This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 81484ac7bcc [fix](file cache) Keep only weak references to file blocks
in CachedRemoteFileReader (#69705)
81484ac7bcc is described below
commit 81484ac7bcc6db9a03529b09afd87994f0f1e92f
Author: Xin Liao <[email protected]>
AuthorDate: Wed Oct 28 18:01:26 2026 +0800
[fix](file cache) Keep only weak references to file blocks in
CachedRemoteFileReader (#69705)
### What problem does this PR solve?
Issue Number: None
Related PR: #68348
Problem Summary:
`CachedRemoteFileReader` keeps the downloaded blocks of its file in
`_cache_file_readers` so later reads can skip `get_or_set()`. Those were
strong references, so `FileBlockCell::releasable()` had to discount them
with a single `_owned_by_cached_reader` flag on the block:
```cpp
use_count == 1 || (use_count == 2 && _owned_by_cached_reader)
```
A flag cannot count readers, and the same segment file is often opened
by several readers at once (the one in `SegmentCache`, readers that
bypass it, a concurrent cache miss that loads a second instance). That
breaks `releasable()` in both directions:
1. Blocks stay busy forever. With two readers `use_count` is 3. When one
of them is destroyed it clears the flag although the other still holds
the block, leaving `use_count == 2` with the flag off. Either way a sync
`clear` (and LRU eviction) can only mark the block deleting, and
#68348's background recycle cannot reclaim it either, because it relies
on `releasable()` too. The block, its file and its bytes in
`ttl_cache_size` stay until the long-lived reader goes away, which for a
reader in `SegmentCache` means until the tablet is cleaned up. In a
cloud regression run, 899 blocks (929,565,612 bytes of TTL cache)
survived every sync clear for over 12 minutes, failing
`create_table_as_select`, `alter_ttl_1`, `test_ttl_preempt` and
`test_warm_up_cluster_bigsize`.
2. Busy blocks look free. `get_blocks_by_key()` also set the flag for
callers that are not readers (`BlockFileCacheTtlMgr`, cache size
statistics, the file cache HTTP action), and nothing ever cleared it. A
block later held only by a `FileBlocksHolder` of a running query then
counts as releasable, so eviction or clear can delete its file mid-read.
This PR stores `std::weak_ptr` in `_cache_file_readers` and removes the
flag. Readers no longer count as users, so `releasable()` is just
`use_count == 1`, however many readers know about a block. A direct read
locks the weak reference for the duration of the read, which keeps the
block from being deleted under it, and falls back to `get_or_set()` once
the cache has dropped the block. `_insert_file_reader()` now replaces a
stale entry at the same offset instead of keeping it, so direct reads of
that range work again after it is cached anew.
---
be/src/io/cache/block_file_cache.cpp | 1 -
be/src/io/cache/block_file_cache.h | 12 +-
be/src/io/cache/cached_remote_file_reader.cpp | 34 +++--
be/src/io/cache/cached_remote_file_reader.h | 6 +-
be/src/io/cache/file_block.h | 4 -
be/test/io/cache/block_file_cache_test.cpp | 163 ++++++++++++++++++++-
.../cluster/test_warm_up_cluster_batch.groovy | 15 +-
.../cloud_p0/cache/test_file_cache_info.groovy | 8 +-
.../suites/cloud_p0/cache/ttl/alter_ttl_4.groovy | 27 +++-
.../cloud_p0/cache/ttl/test_ttl_lru_evict.groovy | 131 +++++++----------
10 files changed, 283 insertions(+), 118 deletions(-)
diff --git a/be/src/io/cache/block_file_cache.cpp
b/be/src/io/cache/block_file_cache.cpp
index f0b9164d93d..1fd6555b8d1 100644
--- a/be/src/io/cache/block_file_cache.cpp
+++ b/be/src/io/cache/block_file_cache.cpp
@@ -2706,7 +2706,6 @@ std::map<size_t, FileBlockSPtr>
BlockFileCache::get_blocks_by_key(const UInt128W
if (_files.contains(hash)) {
for (auto& [offset, cell] : _files[hash]) {
if (cell.file_block->state() == FileBlock::State::DOWNLOADED) {
- cell.file_block->_owned_by_cached_reader = true;
offset_to_block.emplace(offset, cell.file_block);
}
}
diff --git a/be/src/io/cache/block_file_cache.h
b/be/src/io/cache/block_file_cache.h
index 89c03129b18..d3ef128c91a 100644
--- a/be/src/io/cache/block_file_cache.h
+++ b/be/src/io/cache/block_file_cache.h
@@ -159,13 +159,11 @@ struct FileBlockCell {
.count();
}
- /// Pointer to file block is always hold by the cache itself.
- /// Apart from pointer in cache, it can be hold by cache users, when they
call
- /// getorSet(), but cache users always hold it via FileBlocksHolder.
- bool releasable() const {
- return (file_block.use_count() == 1 ||
- (file_block.use_count() == 2 &&
file_block->_owned_by_cached_reader));
- }
+ /// Pointer to file block is always hold by the cache itself. Every other
strong reference
+ /// means somebody is still using the block. CachedRemoteFileReader keeps
only weak references
+ /// in its direct-read map, so however many readers know about a block,
they do not keep it
+ /// busy.
+ bool releasable() const { return file_block.use_count() == 1; }
size_t size() const { return file_block->_block_range.size(); }
diff --git a/be/src/io/cache/cached_remote_file_reader.cpp
b/be/src/io/cache/cached_remote_file_reader.cpp
index 18eb7c3d00d..dc9a8eede5f 100644
--- a/be/src/io/cache/cached_remote_file_reader.cpp
+++ b/be/src/io/cache/cached_remote_file_reader.cpp
@@ -137,7 +137,9 @@ void CachedRemoteFileReader::_init_doris_table_cache() {
_cache = FileCacheFactory::instance()->get_by_path(_cache_hash);
if (_can_read_cache_file_directly()) {
// this is designed for and test in doris table, external table need
extra tests
- _cache_file_readers = _cache->get_blocks_by_key(_cache_hash);
+ for (auto& [offset, block] : _cache->get_blocks_by_key(_cache_hash)) {
+ _cache_file_readers.emplace(offset, block);
+ }
}
}
@@ -179,15 +181,13 @@ void
CachedRemoteFileReader::_insert_file_reader(FileBlockSPtr file_block) {
if (_can_read_cache_file_directly()) {
std::lock_guard lock(_mtx);
DCHECK(file_block->state() == FileBlock::State::DOWNLOADED);
- file_block->_owned_by_cached_reader = true;
- _cache_file_readers.emplace(file_block->offset(),
std::move(file_block));
+ // Overwrite rather than emplace: an older entry at this offset may
point to a block the
+ // cache has already dropped, and keeping it would disable direct
reads of this range.
+ _cache_file_readers.insert_or_assign(file_block->offset(),
std::move(file_block));
}
}
CachedRemoteFileReader::~CachedRemoteFileReader() {
- for (auto& it : _cache_file_readers) {
- it.second->_owned_by_cached_reader = false;
- }
static_cast<void>(close());
}
@@ -777,27 +777,33 @@ bool
CachedRemoteFileReader::_try_read_from_cached_files_directly(
size_t current_offset = offset;
while (need_read_size != 0 && iter != _cache_file_readers.end()) {
- if (iter->second->offset() > current_offset ||
- iter->second->range().right < current_offset) {
+ // Holding the block while reading it makes the cache treat it as
busy, so its file is not
+ // deleted mid-read. If the block is marked deleting meanwhile,
run_background_gc()
+ // reclaims it once this reference is gone.
+ FileBlockSPtr block = iter->second.lock();
+ if (block == nullptr) {
+ // The cache has dropped this block. Read the range through the
cache instead.
+ break;
+ }
+ if (block->offset() > current_offset || block->range().right <
current_offset) {
break;
}
- size_t file_offset = current_offset - iter->second->offset();
- size_t reserve_bytes = std::min(need_read_size,
iter->second->range().size() - file_offset);
+ size_t file_offset = current_offset - block->offset();
+ size_t reserve_bytes = std::min(need_read_size, block->range().size()
- file_offset);
if (is_dryrun) [[unlikely]] {
g_skip_local_cache_io_sum_bytes << reserve_bytes;
} else {
SCOPED_RAW_TIMER(&stats.local_read_timer);
- if (!iter->second
- ->read(Slice(result.data + (current_offset - offset),
reserve_bytes),
- file_offset)
+ if (!block->read(Slice(result.data + (current_offset - offset),
reserve_bytes),
+ file_offset)
.ok()) { // TODO: maybe read failed because block
evict, should handle error
break;
}
source_read_breakdown.local_bytes += reserve_bytes;
}
- _cache->add_need_update_lru_block(iter->second);
+ _cache->add_need_update_lru_block(std::move(block));
need_read_size -= reserve_bytes;
current_offset += reserve_bytes;
already_read += reserve_bytes;
diff --git a/be/src/io/cache/cached_remote_file_reader.h
b/be/src/io/cache/cached_remote_file_reader.h
index 68b112d9ece..1e58a1e4c52 100644
--- a/be/src/io/cache/cached_remote_file_reader.h
+++ b/be/src/io/cache/cached_remote_file_reader.h
@@ -435,7 +435,11 @@ private:
UInt128Wrapper _cache_hash;
BlockFileCache* _cache = nullptr;
std::shared_mutex _mtx;
- std::map<size_t, FileBlockSPtr> _cache_file_readers;
+ // Downloaded blocks this reader may read directly, keyed by offset. The
references are weak on
+ // purpose: the cache decides when a block may go by counting strong
references, and a reader
+ // that lives as long as its segment must not keep a block busy. Several
readers of the same
+ // file can know about the same block at once.
+ std::map<size_t, std::weak_ptr<FileBlock>> _cache_file_readers;
};
} // namespace doris::io
diff --git a/be/src/io/cache/file_block.h b/be/src/io/cache/file_block.h
index 3e29f8ddb6a..a5e0eb875dd 100644
--- a/be/src/io/cache/file_block.h
+++ b/be/src/io/cache/file_block.h
@@ -151,10 +151,6 @@ public:
void set_deleting() { _is_deleting = true; }
bool is_deleting() const { return _is_deleting; };
-public:
- std::atomic<bool> _owned_by_cached_reader {
- false}; // pocessed by CachedRemoteFileReader::_cache_file_readers
-
private:
enum class CacheReferenceRole {
HOLDER,
diff --git a/be/test/io/cache/block_file_cache_test.cpp
b/be/test/io/cache/block_file_cache_test.cpp
index 549efd6b67e..4976fa7020e 100644
--- a/be/test/io/cache/block_file_cache_test.cpp
+++ b/be/test/io/cache/block_file_cache_test.cpp
@@ -3901,7 +3901,7 @@ TEST_F(BlockFileCacheTest,
recycle_deleting_blocks_reclaims_stranded_busy_blocks
auto key = io::BlockFileCache::hash("recycle_deleting_blocks");
// A plain shared_ptr, standing in for the reference holders that do not
release through
- // FileBlock::release_cache_reference(): the background LRU update queue,
the block map of
+ // FileBlock::release_cache_reference(): the background LRU update queue,
the direct read of
// CachedRemoteFileReader, the block vector of BlockFileCacheTtlMgr.
std::vector<io::FileBlockSPtr> busy_refs;
{
@@ -10534,6 +10534,167 @@ TEST_F(BlockFileCacheTest,
EXPECT_TRUE(dryrun_stats.hit_cache);
}
+namespace {
+
+// Downloads the first two tiny blocks of key and, if reader is given,
registers them for its
+// direct reads.
+void cache_first_two_blocks(io::BlockFileCache* cache, const UInt128Wrapper&
key,
+ const std::string& content,
CachedRemoteFileReader* reader) {
+ ReadStatistics cache_stats;
+ auto context = create_cached_remote_reader_context(&cache_stats);
+ FileBlocksHolder holder = cache->get_or_set(key, 0, 8, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 2);
+ for (auto& block : blocks) {
+ append_cached_remote_reader_block(block, content);
+ if (reader != nullptr) {
+ reader->_insert_file_reader(block);
+ }
+ }
+}
+
+void expect_first_two_blocks_releasable(io::BlockFileCache* cache, const
UInt128Wrapper& key) {
+ std::lock_guard<std::mutex> cache_lock(cache->_mutex);
+ for (size_t offset : {0, 4}) {
+ auto* cell = cache->get_cell(key, offset, cache_lock);
+ ASSERT_NE(cell, nullptr) << "offset=" << offset;
+ EXPECT_TRUE(cell->releasable()) << "offset=" << offset;
+ }
+}
+
+} // namespace
+
+// Several readers of one segment file, such as the one in SegmentCache and a
temporary one that
+// bypasses it, know about the same downloaded blocks. None of them may keep
those blocks busy, or
+// a sync clear can only mark the blocks deleting and they stay on disk while
any reader lives.
+TEST_F(BlockFileCacheTest,
cached_remote_file_readers_of_same_file_do_not_keep_blocks_busy) {
+ const bool old_enable_direct = config::enable_read_cache_file_directly;
+ config::enable_read_cache_file_directly = true;
+ Defer restore {[&]() { config::enable_read_cache_file_directly =
old_enable_direct; }};
+
+ const std::string content = "abcdefghijklmnop";
+ const fs::path file_path =
+
create_cached_remote_reader_test_file("cached_remote_readers_not_busy",
content);
+ Defer cleanup_file {[&]() {
+ std::error_code ec;
+ fs::remove(file_path, ec);
+ }};
+
+ const fs::path cache_path = caches_dir /
"cached_remote_readers_not_busy_cache";
+ clear_cached_remote_reader_factory();
+ Defer cleanup_cache {[&]() {
+ std::error_code ignore;
+ fs::remove_all(cache_path, ignore);
+ clear_cached_remote_reader_factory();
+ }};
+
+ auto* cache =
+ create_cached_remote_reader_test_cache(cache_path,
kCachedRemoteReaderTinyBlockSize);
+
+ // A reader of a doris table finds the cached blocks by the file name.
+ const auto key = io::BlockFileCache::hash(file_path.filename().native());
+ cache_first_two_blocks(cache, key, content, nullptr);
+
+ io::FileReaderOptions opts;
+ opts.cache_type = io::FileCachePolicy::FILE_BLOCK_CACHE;
+ opts.is_doris_table = true;
+ opts.tablet_id = 10086;
+ // Each reader closes its own local reader when it goes away.
+ FileReaderSPtr long_lived_local_reader;
+ FileReaderSPtr temporary_local_reader;
+ ASSERT_TRUE(global_local_filesystem()
+ ->open_file(file_path.string(),
&long_lived_local_reader)
+ .ok());
+ ASSERT_TRUE(
+ global_local_filesystem()->open_file(file_path.string(),
&temporary_local_reader).ok());
+ auto long_lived_reader =
+ std::make_shared<CachedRemoteFileReader>(long_lived_local_reader,
opts);
+ auto temporary_reader =
std::make_shared<CachedRemoteFileReader>(temporary_local_reader, opts);
+ ASSERT_EQ(long_lived_reader->_cache_file_readers.size(), 2);
+ ASSERT_EQ(temporary_reader->_cache_file_readers.size(), 2);
+ expect_first_two_blocks_releasable(cache, key);
+
+ // The temporary reader going away must not change how the other one is
accounted for.
+ temporary_reader.reset();
+ expect_first_two_blocks_releasable(cache, key);
+
+ auto msg = cache->clear_file_cache_sync();
+ EXPECT_NE(msg.find("num_cells_to_delete=2"), std::string::npos) << msg;
+ EXPECT_NE(msg.find("num_cells_wait_recycle=0"), std::string::npos) << msg;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache->_mutex);
+ EXPECT_EQ(cache->get_cell(key, 0, cache_lock), nullptr);
+ EXPECT_EQ(cache->get_cell(key, 4, cache_lock), nullptr);
+ }
+
+ // The reader still has entries for the dropped blocks, but must not read
through them.
+ std::string buffer(6, '#');
+ size_t bytes_read = 0;
+ size_t already_read = 0;
+ ReadStatistics read_stats;
+ SourceReadBreakdown source_read_breakdown;
+ EXPECT_FALSE(long_lived_reader->_try_read_from_cached_files_directly(
+ 1, Slice(buffer.data(), buffer.size()), buffer.size(), false,
read_stats,
+ source_read_breakdown, already_read, &bytes_read));
+ EXPECT_EQ(already_read, 0);
+ EXPECT_EQ(buffer, "######");
+
+ // Once the range is cached again, the new blocks replace the stale
entries.
+ cache_first_two_blocks(cache, key, content, long_lived_reader.get());
+ expect_first_two_blocks_releasable(cache, key);
+ EXPECT_TRUE(long_lived_reader->_try_read_from_cached_files_directly(
+ 1, Slice(buffer.data(), buffer.size()), buffer.size(), false,
read_stats,
+ source_read_breakdown, already_read, &bytes_read));
+ EXPECT_EQ(buffer, "bcdefg");
+}
+
+// Looking blocks up by key must not change whether the cache considers them
in use. A block held
+// by a reader through FileBlocksHolder must stay busy, so clear can only mark
it deleting.
+TEST_F(BlockFileCacheTest,
get_blocks_by_key_does_not_make_busy_blocks_releasable) {
+ const std::string content = "abcd";
+ const fs::path cache_path = caches_dir / "get_blocks_by_key_busy_cache";
+ clear_cached_remote_reader_factory();
+ Defer cleanup_cache {[&]() {
+ std::error_code ignore;
+ fs::remove_all(cache_path, ignore);
+ clear_cached_remote_reader_factory();
+ }};
+
+ auto* cache =
+ create_cached_remote_reader_test_cache(cache_path,
kCachedRemoteReaderTinyBlockSize);
+ const auto key = io::BlockFileCache::hash("get_blocks_by_key_busy");
+ ReadStatistics cache_stats;
+ auto context = create_cached_remote_reader_context(&cache_stats);
+
+ auto holder = std::make_unique<FileBlocksHolder>(cache->get_or_set(key, 0,
4, context));
+ ASSERT_EQ(holder->file_blocks.size(), 1);
+ append_cached_remote_reader_block(holder->file_blocks.front(), content);
+
+ ASSERT_EQ(cache->get_blocks_by_key(key).size(), 1);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache->_mutex);
+ auto* cell = cache->get_cell(key, 0, cache_lock);
+ ASSERT_NE(cell, nullptr);
+ EXPECT_FALSE(cell->releasable());
+ }
+
+ auto msg = cache->clear_file_cache_sync();
+ EXPECT_NE(msg.find("num_cells_wait_recycle=1"), std::string::npos) << msg;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache->_mutex);
+ auto* cell = cache->get_cell(key, 0, cache_lock);
+ ASSERT_NE(cell, nullptr);
+ EXPECT_TRUE(cell->file_block->is_deleting());
+ }
+
+ // The holder is the last user, so releasing it finishes the deletion.
+ holder.reset();
+ {
+ std::lock_guard<std::mutex> cache_lock(cache->_mutex);
+ EXPECT_EQ(cache->get_cell(key, 0, cache_lock), nullptr);
+ }
+}
+
TEST_F(BlockFileCacheTest,
file_block_appendv_persists_segmented_payload_to_fs_cache) {
const fs::path cache_path = caches_dir / "file_block_appendv_fs_cache";
clear_cached_remote_reader_factory();
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_batch.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_batch.groovy
index 9244e7f9e24..d31750bef02 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_batch.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_batch.groovy
@@ -211,11 +211,16 @@ suite("test_warm_up_cluster_batch", "nonConcurrent") {
}
int batch = (round + 1) / 2
def hotspotBeforeQueries = getHotspotCounts()
- // One real scan per table is enough to advance the hotspot
counters for this batch.
- tableNames.each { tableName ->
- def result = sql(dataQueries[tableName])
- assertTrue(!result.isEmpty() && result[0].every { it !=
null },
- "Data query should read nonempty ${tableName}")
+ // After its first report, BE reports a partition again only
when its queries per
+ // day grew by at least 5 or its last access moved by at least
60 seconds since that
+ // report (get_return_partitions() in
cloud_tablet_hotspot.cpp). Each scan adds one,
+ // and batches can be less than 60 seconds apart, so scan each
table 5 times.
+ for (int i = 0; i < 5; i++) {
+ tableNames.each { tableName ->
+ def result = sql(dataQueries[tableName])
+ assertTrue(!result.isEmpty() && result[0].every { it
!= null },
+ "Data query should read nonempty ${tableName}")
+ }
}
// Require hotspot counters for these new table/partition IDs
to advance beyond
// the pre-query snapshot, without assuming identical FE and
BE timestamps.
diff --git a/regression-test/suites/cloud_p0/cache/test_file_cache_info.groovy
b/regression-test/suites/cloud_p0/cache/test_file_cache_info.groovy
index 26c4964bb72..51034cce2ed 100644
--- a/regression-test/suites/cloud_p0/cache/test_file_cache_info.groovy
+++ b/regression-test/suites/cloud_p0/cache/test_file_cache_info.groovy
@@ -37,7 +37,7 @@ suite("test_file_cache_info", "nonConcurrent") {
Long tabletId = null
def lastState = [:]
def clearResponses = [:]
- def originalQueryCache = sql("select @@enable_sql_cache,
@@enable_query_cache")[0].collect { value ->
+ def originalQueryCache = sql("select @@enable_sql_cache,
@@enable_query_cache, @@enable_page_cache")[0].collect { value ->
String setting = value.toString().toLowerCase(Locale.ROOT)
assertTrue(setting in ["true", "false", "0", "1"], "Unexpected
cache setting: ${value}")
setting
@@ -168,6 +168,9 @@ suite("test_file_cache_info", "nonConcurrent") {
sql "use @${clusterName}"
sql "set enable_sql_cache = false"
sql "set enable_query_cache = false"
+ // The reread after clear must go through the file cache. Pages of
a five-row table
+ // stay in the BE page cache, and serving them from memory would
never refill it.
+ sql "set enable_page_cache = false"
sql "drop table if exists ${tableName} force"
sql """CREATE TABLE ${tableName} (
c_custkey INT, c_name STRING, c_address STRING, c_city STRING,
@@ -214,7 +217,8 @@ suite("test_file_cache_info", "nonConcurrent") {
} finally {
def cleanupErrors = []
try { sql "drop table if exists ${tableName} force" } catch
(Throwable t) { cleanupErrors.add(t) }
- [enable_sql_cache: originalQueryCache[0], enable_query_cache:
originalQueryCache[1]].each { name, value ->
+ [enable_sql_cache: originalQueryCache[0], enable_query_cache:
originalQueryCache[1],
+ enable_page_cache: originalQueryCache[2]].each { name, value ->
try { sql "set ${name} = ${value}" } catch (Throwable t) {
cleanupErrors.add(t) }
}
if (!cleanupErrors.isEmpty()) {
diff --git a/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_4.groovy
b/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_4.groovy
index c78a140ba47..3ec3864f6f5 100644
--- a/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_4.groovy
+++ b/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_4.groovy
@@ -42,7 +42,7 @@ suite("alter_ttl_4", "nonConcurrent") {
def pendingLoads = [] as Set
def lastState = [:]
def cacheDifference = [:]
- def originalQueryCache = sql("select @@enable_sql_cache,
@@enable_query_cache")[0].collect { value ->
+ def originalQueryCache = sql("select @@enable_sql_cache,
@@enable_query_cache, @@enable_page_cache")[0].collect { value ->
String setting = value.toString().toLowerCase(Locale.ROOT)
assertTrue(setting in ["true", "false", "0", "1"], "Unexpected
cache setting: ${value}")
setting
@@ -206,6 +206,8 @@ suite("alter_ttl_4", "nonConcurrent") {
sql "use @${clusterName}"
sql "set enable_sql_cache = false"
sql "set enable_query_cache = false"
+ // Scans of the evicted target must read its files again instead
of pages kept in memory.
+ sql "set enable_page_cache = false"
[(targetTable): initialTtl, (pressureTable): 0L].each { table, ttl
->
sql "drop table if exists ${table} force"
String ddl = new
File("${context.file.parent}/../ddl/customer_ttl.sql").text
@@ -291,14 +293,22 @@ suite("alter_ttl_4", "nonConcurrent") {
logMetricDelta("Pressure", metricsBeforePressure,
metricsAfterPressure)
assertTrue(!evictedFiles.isEmpty(), "No target cache eviction
before pressure limit: " +
"imported_rows=${pressureRows}/${maxPressureRows},
metrics=${metricsAfterPressure}")
+ def coverageAfterPressure = coverage(getCache())
assertEquals(dataBaseline, scanTarget())
- // Re-reading an evicted file may split blocks differently.
Compare merged byte ranges.
- def reloaded = waitForCache("reloaded NORMAL cache coverage",
"normal") { blocks ->
- def current = coverage(blocks)
- cacheDifference = [missing: missingCoverage(baselineCoverage,
current).take(10),
- extra: missingCoverage(current,
baselineCoverage).take(10)]
- current == baselineCoverage
+ // The scan reads the data pages of the evicted files again, but
not their footers and
+ // indexes, which stay in memory with the open segments. So the
reloaded ranges need not
+ // match what the load cached: require the evicted files to have
more cached than right
+ // after the eviction.
+ def cachedBytes = { Map fileRanges ->
+ evictedFiles.keySet().sum(0L) { file -> (fileRanges[file] ?:
[]).sum(0L) { it[1] - it[0] } }
+ }
+ long bytesAfterPressure = cachedBytes(coverageAfterPressure)
+ def reloaded = waitForCache("reloaded NORMAL cache of evicted
files", "normal") { blocks ->
+ long bytesNow = cachedBytes(coverage(blocks))
+ cacheDifference = [evicted_files: evictedFiles.size(),
+ cached_bytes_after_pressure:
bytesAfterPressure, cached_bytes_now: bytesNow]
+ bytesNow > bytesAfterPressure
}
def metricsBeforeExtension = getMetrics()
long now = sql("select unix_timestamp()")[0][0] as Long
@@ -322,7 +332,8 @@ suite("alter_ttl_4", "nonConcurrent") {
[targetTable, pressureTable].each { table ->
try { sql "drop table if exists ${table} force" } catch
(Throwable t) { cleanupErrors.add(t) }
}
- [enable_sql_cache: originalQueryCache[0], enable_query_cache:
originalQueryCache[1]].each { name, value ->
+ [enable_sql_cache: originalQueryCache[0], enable_query_cache:
originalQueryCache[1],
+ enable_page_cache: originalQueryCache[2]].each { name, value ->
try { sql "set ${name} = ${value}" } catch (Throwable t) {
cleanupErrors.add(t) }
}
if (!cleanupErrors.isEmpty()) {
diff --git
a/regression-test/suites/cloud_p0/cache/ttl/test_ttl_lru_evict.groovy
b/regression-test/suites/cloud_p0/cache/ttl/test_ttl_lru_evict.groovy
index f452d071e04..e69e59f2e6a 100644
--- a/regression-test/suites/cloud_p0/cache/ttl/test_ttl_lru_evict.groovy
+++ b/regression-test/suites/cloud_p0/cache/ttl/test_ttl_lru_evict.groovy
@@ -58,7 +58,10 @@ suite("test_ttl_lru_evict") {
logger.info("getS3SK:${getS3SK()}");
logger.info("getS3Endpoint:${getS3Endpoint()}");
- def ttlProperties = """ PROPERTIES("file_cache_ttl_seconds"="150") """
+ // Long enough that the first loads are still TTL when the TTL queue
overflows. How many loads
+ // that takes depends on the cache capacity of the environment.
+ def ttlSeconds = 300
+ def ttlProperties = """
PROPERTIES("file_cache_ttl_seconds"="${ttlSeconds}") """
String[][] backends = sql """ show backends """
String backendId;
def backendIdToBackendIP = [:]
@@ -206,65 +209,53 @@ suite("test_ttl_lru_evict") {
}
}
- clearFileCache.call() {
- respCode, body -> {}
- }
-
- long ttl_cache_evict_size_begin = 0;
- getMetricsMethod.call() {
- respCode, body ->
- assertEquals("${respCode}".toString(), "200")
- String out = "${body}".toString()
- def strs = out.split('\n')
- Boolean flag1 = false;
- for (String line in strs) {
- if (flag1) break;
- if (line.contains("ttl_cache_evict_size")) {
- if (line.startsWith("#")) {
+ // Values of every metric line whose name contains the given name, one per
cache path.
+ def getMetricValues = { String name ->
+ def values = []
+ getMetricsMethod.call() {
+ respCode, body ->
+ assertEquals("${respCode}".toString(), "200")
+ for (String line in "${body}".toString().split('\n')) {
+ if (line.startsWith("#") || !line.contains(name)) {
continue
}
- def i = line.indexOf(' ')
- ttl_cache_evict_size_begin = line.substring(i).toLong()
- flag1 = true
+ values.add(line.substring(line.indexOf('
')).trim().toLong())
}
- }
- assertTrue(flag1)
+ }
+ assertTrue(!values.isEmpty(), "metric ${name} not found")
+ return values
}
- // load for MANY times to this dup table to make cache full enough to evict
- // It will takes huge amount of time, so it should be p1/p2 level case.
- // But someone told me it is the right place. Never mind just do it!
+ clearFileCache.call() {
+ respCode, body -> {}
+ }
- for (int i = 0; i < 10; i++) {
+ long ttl_cache_evict_size_begin =
getMetricValues("ttl_cache_evict_size")[0]
+ long ttl_cache_size_begin = getMetricValues("ttl_cache_size")[0]
+ long cache_capacity = getMetricValues("file_cache_capacity").sum()
+
+ // first, we test whether ttl evict actively.
+ // Load to this dup table until the TTL queue overflows. A fixed number of
loads only
+ // overflows it on a small enough cache, so keep loading, bounded by the
cache capacity.
+ long evicted = 0
+ long loads = 0
+ long max_loads = 10
+ while (evicted <= 238860800 && loads < max_loads) {
load_customer_once("customer_ttl")
+ loads++
+ if (loads == 1) {
+ long bytes_per_load = getMetricValues("ttl_cache_size")[0] -
ttl_cache_size_begin
+ assertTrue(bytes_per_load > 0, "the first load did not add any TTL
cache")
+ // The TTL queue is part of the capacity, so this many loads
overflow it for sure.
+ max_loads = Math.max(max_loads,
cache_capacity.intdiv(bytes_per_load) + 2)
+ logger.info("cache capacity: ${cache_capacity}, TTL bytes per
load: ${bytes_per_load}, " +
+ "max loads: ${max_loads}")
+ }
+ evicted = getMetricValues("ttl_cache_evict_size")[0] -
ttl_cache_evict_size_begin
+ logger.info("ttl evict diff after ${loads} loads: ${evicted}")
}
- sleep(10000) // 10s
- // first, we test whether ttl evict actively
- long ttl_cache_evict_size_end = 0;
- getMetricsMethod.call() {
- respCode, body ->
- assertEquals("${respCode}".toString(), "200")
- String out = "${body}".toString()
- def strs = out.split('\n')
- Boolean flag1 = false;
- for (String line in strs) {
- if (flag1) break;
- if (line.contains("ttl_cache_evict_size")) {
- if (line.startsWith("#")) {
- continue
- }
- def i = line.indexOf(' ')
- ttl_cache_evict_size_end = line.substring(i).toLong()
- flag1 = true
- }
- }
- assertTrue(flag1)
- }
- // Note: this only applies when the case is run
- // sequentially, coz we don't know what other cases are
- // doing with TTL cache
- logger.info("ttl evict diff:" + (ttl_cache_evict_size_end -
ttl_cache_evict_size_begin).toString())
- assertTrue((ttl_cache_evict_size_end - ttl_cache_evict_size_begin) >
238860800)
+ assertTrue(evicted > 238860800,
+ "TTL cache evicted only ${evicted} bytes after ${loads} loads,
capacity ${cache_capacity}")
// then we test skip_cache count when doing query when ttl cache is full
// we expect it to be rare
@@ -324,29 +315,19 @@ suite("test_ttl_lru_evict") {
// finally, we test whether LRU queue clean itself up when all ttl
// cache evict after expiration
- sleep(200000)
- getMetricsMethod.call() {
- respCode, body ->
- assertEquals("${respCode}".toString(), "200")
- String out = "${body}".toString()
- def strs = out.split('\n')
- Boolean flag1 = false;
- for (String line in strs) {
- if (flag1) break;
- if (line.contains("ttl_cache_lru_queue_size")) {
- if (line.startsWith("#")) {
- continue
- }
- def i = line.indexOf(' ')
- // all ttl will expire, so the ttl LRU queue set to 0
- // Note: this only applies when the case is run
- // sequentially, coz we don't know what other cases are
- // doing with TTL cache
- assertEquals(line.substring(i).toLong(), 0)
- flag1 = true
- }
- }
- assertTrue(flag1)
+ // Note: this only applies when the case is run
+ // sequentially, coz we don't know what other cases are
+ // doing with TTL cache
+ long ttl_queue_size = -1
+ long deadline = System.currentTimeMillis() + (ttlSeconds + 120) * 1000L
+ while (System.currentTimeMillis() < deadline) {
+ // all ttl will expire, so the ttl LRU queue set to 0
+ ttl_queue_size = getMetricValues("ttl_cache_lru_queue_size")[0]
+ if (ttl_queue_size == 0) {
+ break
+ }
+ sleep(10000)
}
+ assertEquals(0, ttl_queue_size)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
