gavinchou commented on code in PR #60001:
URL: https://github.com/apache/doris/pull/60001#discussion_r2702538329
##########
be/src/cloud/cloud_tablet.cpp:
##########
@@ -1736,5 +1557,212 @@ void CloudTablet::add_warmed_up_rowset(const RowsetId&
rowset_id) {
.start_tp = std::chrono::steady_clock::now()};
}
+bool CloudTablet::_check_rowset_should_be_visible_but_not_warmed_up(
+ const RowsetMetaSharedPtr& rs_meta, int64_t path_max_version,
+ std::chrono::system_clock::time_point freshness_limit_tp) const {
+ if (rs_meta->version() == Version {0, 1}) {
+ // skip rowset[0-1]
+ return false;
+ }
+ bool ret = rs_meta->start_version() > path_max_version &&
+ rs_meta->visible_timestamp() < freshness_limit_tp;
+ if (ret && config::read_cluster_cache_opt_verbose_log) {
+ using namespace std::chrono;
+ std::time_t t1 = system_clock::to_time_t(rs_meta->visible_timestamp());
+ std::tm tm1 = *std::localtime(&t1);
+ std::ostringstream oss1;
+ oss1 << std::put_time(&tm1, "%Y-%m-%d %H:%M:%S");
+
+ std::time_t t2 = system_clock::to_time_t(freshness_limit_tp);
+ std::tm tm2 = *std::localtime(&t2);
+ std::ostringstream oss2;
+ oss2 << std::put_time(&tm2, "%Y-%m-%d %H:%M:%S");
+ LOG_INFO(
+ "[verbose]
CloudTablet::capture_rs_readers_with_freshness_tolerance, "
+ "find a rowset which should be visible but not warmed up,
tablet_id={}, "
+ "path_max_version={}, rowset_id={}, version={},
visible_time={}, "
+ "freshness_limit={}, version_graph={},
rowset_warmup_digest={}",
+ tablet_id(), path_max_version,
rs_meta->rowset_id().to_string(),
+ rs_meta->version().to_string(), oss1.str(), oss2.str(),
+ _timestamped_version_tracker.debug_string(),
rowset_warmup_digest());
+ }
+ return ret;
+}
+
+void CloudTablet::_submit_segment_download_task(const RowsetSharedPtr& rs, int
seg_id,
+ const RowsetMetaSharedPtr&
rowset_meta,
+ int64_t expiration_time) {
+ auto self = std::dynamic_pointer_cast<CloudTablet>(shared_from_this());
+ auto storage_resource = rowset_meta->remote_storage_resource();
+ if (!storage_resource) {
+ LOG(WARNING) << storage_resource.error();
+ return;
+ }
+
+ // clang-format off
+
_engine.file_cache_block_downloader().submit_download_task(io::DownloadFileMeta
{
+ .path =
storage_resource.value()->remote_segment_path(*rowset_meta, seg_id),
+ .file_size = rs->rowset_meta()->segment_file_size(seg_id),
+ .file_system = storage_resource.value()->fs,
+ .ctx = {
+ .expiration_time = expiration_time,
+ .is_dryrun =
config::enable_reader_dryrun_when_download_file_cache,
+ .is_warmup = true
+ },
+ .download_done {[=](Status st) {
+
DBUG_EXECUTE_IF("CloudTablet::add_rowsets.download_data.callback.block_compaction_rowset",
{
+ if (rs->version().second > rs->version().first) {
+ auto sleep_time = dp->param<int>("sleep", 3);
+ LOG_INFO(
+ "[verbose] block download for
rowset={}, "
+ "version={}, sleep={}",
+ rs->rowset_id().to_string(),
+ rs->version().to_string(), sleep_time);
+ std::this_thread::sleep_for(
+ std::chrono::seconds(sleep_time));
+ }
+ });
+
self->complete_rowset_segment_warmup(WarmUpTriggerSource::SYNC_ROWSET,
rowset_meta->rowset_id(), st, 1, 0);
+ if (!st) {
+ LOG_WARNING("add rowset warm up error ").error(st);
+ }
+ }},
+ });
+ // clang-format on
+}
+
+void CloudTablet::_submit_inverted_index_download_task(const RowsetSharedPtr&
rs,
+ const
RowsetMetaSharedPtr& rowset_meta,
+ const io::Path&
idx_path, int64_t idx_size,
+ int64_t
expiration_time) {
+ auto self = std::dynamic_pointer_cast<CloudTablet>(shared_from_this());
+ auto storage_resource = rowset_meta->remote_storage_resource();
+ if (!storage_resource) {
+ LOG(WARNING) << storage_resource.error();
+ return;
+ }
+
+ io::DownloadFileMeta meta {
+ .path = idx_path,
+ .file_size = idx_size,
+ .file_system = storage_resource.value()->fs,
+ .ctx = {.expiration_time = expiration_time,
+ .is_dryrun =
config::enable_reader_dryrun_when_download_file_cache,
+ .is_warmup = true},
+ .download_done {[=](Status st) {
+
DBUG_EXECUTE_IF("CloudTablet::add_rowsets.download_idx.callback.block", {
+ // clang-format on
+ auto sleep_time = dp->param<int>("sleep", 3);
+ LOG_INFO(
+ "[verbose] block download for "
+ "rowset={}, inverted_idx_file={}, "
+ "sleep={}",
+ rs->rowset_id().to_string(), idx_path.string(),
sleep_time);
+
std::this_thread::sleep_for(std::chrono::seconds(sleep_time));
+ // clang-format off
+ });
+
self->complete_rowset_segment_warmup(WarmUpTriggerSource::SYNC_ROWSET,
rowset_meta->rowset_id(), st, 0, 1);
+ if (!st) {
+ LOG_WARNING("add rowset warm up error ").error(st);
+ }
+ }},
+ };
+
self->update_rowset_warmup_state_inverted_idx_num_unlocked(WarmUpTriggerSource::SYNC_ROWSET,
rowset_meta->rowset_id(), 1);
+
_engine.file_cache_block_downloader().submit_download_task(std::move(meta));
+ g_file_cache_cloud_tablet_submitted_index_num << 1;
+ g_file_cache_cloud_tablet_submitted_index_size << idx_size;
+}
+
+void CloudTablet::_add_rowsets_directly(std::vector<RowsetSharedPtr>& rowsets,
bool warmup_delta_data) {
+ for (auto& rs : rowsets) {
+ if (warmup_delta_data) {
+#ifndef BE_TEST
Review Comment:
try this to get rid of large block of `ifndef`
```suggestion
#ifdef BE_TEST
warmup_delta_data = false;
#endif
for (auto& rs : rowsets) {
if (warmup_delta_data) {
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
