liaoxin01 commented on code in PR #68991: URL: https://github.com/apache/doris/pull/68991#discussion_r4273520513
########## be/src/io/cache/tablet_cache_policy_registry.cpp: ########## @@ -0,0 +1,437 @@ +// 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 "io/cache/tablet_cache_policy_registry.h" + +#include <bvar/bvar.h> + +#include <algorithm> +#include <chrono> +#include <limits> + +#include "common/config.h" +#include "common/logging.h" +#include "cpp/sync_point.h" +#include "io/cache/cache_block_meta_store.h" +#include "runtime/exec_env.h" +#include "storage/tablet/tablet_meta.h" +#include "util/thread.h" +#include "util/time.h" + +namespace doris::io { + +bvar::Adder<uint64_t> g_tablet_cache_policy_fetch_num("file_cache_tablet_policy_fetch_num"); +bvar::Adder<uint64_t> g_tablet_cache_policy_fetch_failed_num( + "file_cache_tablet_policy_fetch_failed_num"); +bvar::PassiveStatus<size_t> g_tablet_cache_policy_num( + "file_cache_tablet_policy_num", + [](void*) { return TabletCachePolicyRegistry::instance()->size(); }, nullptr); + +namespace { + +// How often the record is scanned for policies nobody needs any more. +constexpr int64_t kPruneIntervalS = 60; +// How long a tablet whose policy could not be fetched is left alone. +constexpr int64_t kFetchRetryIntervalS = 60; +// How many reported tablets are looked up per hold of the exclusive lock. +constexpr size_t kReportBatchSize = 1024; +// How often the background refresh runs. +constexpr std::chrono::seconds kRefreshInterval {1}; + +Status fetch_from_tablet_meta(int64_t tablet_id, TabletCachePolicy* policy) { + TabletMetaSharedPtr tablet_meta; + RETURN_IF_ERROR(ExecEnv::get_tablet_meta(tablet_id, &tablet_meta, false)); + DORIS_CHECK(tablet_meta != nullptr); + policy->ttl_seconds = tablet_meta->ttl_seconds(); + policy->ctime = tablet_meta->creation_time(); + return Status::OK(); +} + +} // namespace + +int64_t TabletCachePolicy::deadline() const { + if (ttl_seconds <= 0 || ctime <= 0) { + return 0; + } + // Same rule as TabletMeta::file_cache_ttl_expiration_time(): a huge ttl must not wrap into + // a deadline in the past. + if (ctime > std::numeric_limits<int64_t>::max() - ttl_seconds) { + return std::numeric_limits<int64_t>::max(); + } + return ctime + ttl_seconds; +} + +TabletCachePolicyRegistry* TabletCachePolicyRegistry::instance() { + // Never destroyed: a cache path may still be running when the process exits. + static auto* registry = new TabletCachePolicyRegistry(); + return registry; +} + +std::optional<int64_t> TabletCachePolicyRegistry::deadline(int64_t tablet_id) const { + std::shared_lock lock(_mutex); + auto it = _entries.find(tablet_id); + if (it == _entries.end()) { + return std::nullopt; + } + return it->second.policy.deadline(); +} + +int64_t TabletCachePolicyRegistry::protected_until(int64_t tablet_id, int64_t legacy_deadline, + bool* legacy) { + // tablet_id 0 is data outside any Doris tablet (external tables, v2 cache files that never + // recorded one): it only has the deadline it was written with. + if (tablet_id > 0) { + if (const std::optional<int64_t> known = deadline(tablet_id); known.has_value()) { + *legacy = false; + return *known; + } + note_tablet(tablet_id); + } + *legacy = true; + return legacy_deadline; +} + +bool TabletCachePolicyRegistry::is_protected(int64_t tablet_id, int64_t legacy_deadline, + int64_t now) { + bool legacy = false; + return protected_until(tablet_id, legacy_deadline, &legacy) > now; +} + +std::optional<TabletCachePolicy> TabletCachePolicyRegistry::set_locked( + int64_t tablet_id, const TabletCachePolicy& policy, int64_t refreshed_at_s) { + auto [it, inserted] = + _entries.try_emplace(tablet_id, Entry {policy, refreshed_at_s, refreshed_at_s}); + std::optional<TabletCachePolicy> before; + if (!inserted) { + before = it->second.policy; + if (const int64_t old_deadline = before->deadline(); old_deadline > 0) { + _deadlines.erase(_deadlines.find(old_deadline)); + } + it->second.policy = policy; + it->second.refreshed_at_s = refreshed_at_s; + } + if (const int64_t new_deadline = policy.deadline(); new_deadline > 0) { + _deadlines.insert(new_deadline); + } + return before; +} + +std::optional<TabletCachePolicy> TabletCachePolicyRegistry::erase_locked(int64_t tablet_id) { + auto it = _entries.find(tablet_id); + if (it == _entries.end()) { + return std::nullopt; + } + const TabletCachePolicy before = it->second.policy; + if (const int64_t old_deadline = before.deadline(); old_deadline > 0) { + _deadlines.erase(_deadlines.find(old_deadline)); + } + _entries.erase(it); + return before; +} + +void TabletCachePolicyRegistry::bump_versions(const std::optional<TabletCachePolicy>& before, + const std::optional<TabletCachePolicy>& after) { + // A block is protected while the deadline is ahead, so a later deadline can only make blocks + // protected and an earlier one only end their protection. An unknown tablet is judged by the + // deadline each block carries, which may be earlier or later, so learning or forgetting a + // tablet can move its blocks either way. The -1 is never compared: a missing side bumps both. + const int64_t old_deadline = before.has_value() ? before->deadline() : -1; + const int64_t new_deadline = after.has_value() ? after->deadline() : -1; + if (!before.has_value() || !after.has_value() || new_deadline > old_deadline) { Review Comment: Fixed in 4d896bd09da: a change of a known tablet bumps a version only when it protects or unprotects blocks at that moment; a deadline still ahead starts a TTL queue pass when it passes. ########## be/src/io/cache/fs_file_cache_storage.cpp: ########## @@ -155,7 +155,14 @@ Status FSFileCacheStorage::init(BlockFileCache* mgr) { metrics_prefix, "file_cache_leak_removed_files_cnt"); _cache_base_path = mgr->_cache_base_path; _mgr = mgr; - _meta_store = std::make_unique<CacheBlockMetaStore>(_cache_base_path + "/meta", 10000); + _meta_store = std::make_unique<CacheBlockMetaStore>(_cache_base_path + "/meta"); + // The policies known before the restart, so the blocks below are judged by them as they load. + auto* policy_registry = TabletCachePolicyRegistry::instance(); + _meta_store->for_each_tablet_policy( + [policy_registry](int64_t tablet_id, const TabletCachePolicy& policy) { + policy_registry->restore(tablet_id, policy); Review Comment: Not fixing: it needs a downgrade, an ALTER on the older version, then a re-upgrade on the same path, and the restored policy is due for a refresh, so it is corrected within minutes. ########## be/src/io/cache/block_file_cache.cpp: ########## @@ -1402,16 +1416,41 @@ void BlockFileCache::find_evict_candidates(LRUQueue& queue, size_t size, size_t size_t cell_size = cell->size(); DCHECK(entry_size == cell_size); - if (cell->releasable()) { + if (!ttl_queue && is_protected(*cell->file_block, now)) { + // Its tablet gained a TTL after the block was cached and the queue sweep has not + // reached it yet. Never evicted as unprotected data; moved out of the way instead. + if (to_relocate.size() < + static_cast<size_t>(config::file_cache_ttl_relocate_batch_blocks)) { + to_relocate.push_back(cell); + } + if (skip_evict_candidate(skipped)) { + break; + } + } else if (cell->releasable()) { auto& file_block = cell->file_block; std::lock_guard block_lock(file_block->_mutex); DCHECK(file_block->_download_state == FileBlock::State::DOWNLOADED); to_evict.push_back(cell); removed_size += cell_size; cur_removed_size += cell_size; + if (ttl_queue && is_protected(*file_block, now)) { + *_ttl_protected_evicted_bytes << cell_size; Review Comment: Fixed in ccf491d778d: the query-quota path counts protected blocks it evicts too. ########## regression-test/suites/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.groovy: ########## @@ -80,34 +82,77 @@ suite("ex06_ttl_restart_consistency", "docker") { assertTrue(false, "Timeout waiting type=${expectedType}, tablets=${tabletIds}") } - def assertTtlDisabledInShowCreate = { - def rows = sql """show create table ${tableName};""" - assertTrue(!rows.isEmpty(), "No show create table result for ${tableName}") + // TYPE above is computed from the tablet's current TTL. Where the blocks actually are + // shows in the size of the TTL queue, which must come to hold exactly the blocks of + // the given tablets. + def ttlQueueBytes = { + long total = 0 + sql_return_maparray("SHOW BACKENDS").each { be -> + def metrics = new URL("http://${be.Host}:${be.BrpcPort}/brpc_metrics";).text + (metrics =~ /file_cache_ttl_cache_size\s+(\d+)/).each { m -> total += m[1] as long } + } + total + } + def cachedBytes = { List<Long> tabletIds -> + def rows = sql """select sum(size) from information_schema.file_cache_info + where tablet_id in (${tabletIds.join(',')})""" + rows[0][0] == null ? 0L : rows[0][0] as long + } + def waitForTtlQueueToHold = { List<Long> tabletIds, long timeoutMs = 600000L -> + long start = System.currentTimeMillis() + while (System.currentTimeMillis() - start < timeoutMs) { + long expected = cachedBytes(tabletIds) + long actual = ttlQueueBytes() + logger.info("TTL queue holds ${actual} bytes, tablets ${tabletIds} have ${expected}") + if (expected > 0 && actual == expected) { + return + } + sleep(2000) + } + assertTrue(false, "TTL queue does not hold exactly the blocks of tablets ${tabletIds}") + } + + def assertTtlInShowCreate = { String table, long ttlSeconds -> + def rows = sql """show create table ${table};""" + assertTrue(!rows.isEmpty(), "No show create table result for ${table}") def createTableSql = rows[0][1].toString() - logger.info("show create table ${tableName}: ${createTableSql}") - assertTrue(createTableSql.contains("\"file_cache_ttl_seconds\" = \"0\""), - "file_cache_ttl_seconds is not 0 in show create table result: ${createTableSql}") + logger.info("show create table ${table}: ${createTableSql}") + assertTrue(createTableSql.contains("\"file_cache_ttl_seconds\" = \"${ttlSeconds}\""), + "file_cache_ttl_seconds is not ${ttlSeconds} in show create table result: ${createTableSql}") } def values = (0..<300).collect { i -> "(${i}, 'restart_${i}')" }.join(",") sql """insert into ${tableName} values ${values}""" + sql """insert into ${gainTableName} values ${values}""" qt_ex06_preheat """select count(*) from ${tableName} where c1 like 'restart_%'""" + qt_ex06_gain_preheat """select count(*) from ${gainTableName} where c1 like 'restart_%'""" sleep(5000) - def tabletIds = getTabletIds.call() + def tabletIds = getTabletIds.call(tableName) + def gainTabletIds = getTabletIds.call(gainTableName) waitForFileCacheType.call(tabletIds, "ttl") + waitForFileCacheType.call(gainTabletIds, "normal") Review Comment: Fixed in 997e9d7fa16: the TTL cases waiting for NORMAL now accept any type but TTL (ex06, test_ttl, alter_ttl_1/2/3/4, st04). -- 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]
