github-actions[bot] commented on code in PR #68991:
URL: https://github.com/apache/doris/pull/68991#discussion_r4273176079
##########
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:
[P2] Let regression waits accept preserved INDEX blocks
The new gain-table wait assumes every block is NORMAL. With the default
asynchronous flush/packed cache fill, INSERT can finish before the NORMAL fill
starts; the immediately preceding query then reads the segment footer with
`is_index_data=true` and admits an INDEX block. Later writer fills reuse that
cell, whose class is fixed, so this wait times out even after the five-second
sleep. The expiry/removal waits in `test_ttl`, `alter_ttl_2`,
`st04_alter_ttl_n_to_0_runtime` and `alter_ttl_4` similarly still require
NORMAL although INDEX is now an intended result. Accept the preserved non-TTL
class while retaining the block/size and TTL-on checks.
##########
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:
[P2] Count protected evictions in the query-quota path
This is the only increment of `file_cache_ttl_protected_evicted_bytes`, but
`try_reserve()` also LRU-evicts protected blocks directly from the query
context at lines 1529-1545. With query limiting enabled, fill a query's quota
with downloaded TTL blocks, release their holders, then write another protected
block: the oldest is removed without entering this helper, so the counter stays
zero despite evicting protected bytes. Share this accounting with the
query-quota eviction path and add coverage with a nonzero query id and live
query-context holder; the new protected-eviction test only exercises ordinary
cache-capacity eviction.
##########
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:
[P2] Avoid global sweeps for TTL rewrites that stay active
With settled queues, changing a known tablet from TTL 3600 to 7200 while
both deadlines are future changes no block's protection, but this comparison
increments the gained version and starts a full sweep of all three class queues
on every cache path. A still-active shortening similarly scans the whole TTL
queue. Scheduled rewrites therefore cause O(all cached blocks) maintenance
under the cache mutex, even with zero reclassifications; the deleted
`RewritingTtlToAnotherValidValueDoesNotRescanBlocks` test covered this
workload. Bump sweep versions for actual protection transitions between known
policies, retaining deadline tracking for expiry and conservative invalidation
for unknown policies.
##########
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:
[P2] Validate saved policies after an intervening downgrade
A downgrade/re-upgrade can restore an obsolete policy over newer block
metadata: persist `ttl=0` with this version, run the older BE, ALTER the tablet
to TTL 3600 and cache new blocks, then re-upgrade on the same path. The older
BE updates the blocks' future deadlines but never updates or clears the policy
rows in the default column family. This restore makes the old zero
authoritative, so those blocks load as NORMAL and can be evicted before the
first successful metadata fetch. Validate or invalidate policy copies after a
pre-policy writer has used the cache, and cover this compatibility sequence.
This requires neither concurrent policy updates nor a cache-path change.
--
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]
