diff --git a/be/src/cloud/cloud_internal_service.cpp b/be/src/cloud/cloud_internal_service.cpp index 98c6c9790beb69..a423f8622f9f05 100644 --- a/be/src/cloud/cloud_internal_service.cpp +++ b/be/src/cloud/cloud_internal_service.cpp @@ -1202,7 +1202,7 @@ void CloudInternalServiceImpl::warm_up_rowset(google::protobuf::RpcController* c << " us, tablet_id: " << rs_meta.tablet_id() << ", rowset_id: " << rowset_id.to_string(); } - int64_t expiration_time = tablet_meta->ttl_seconds(); + int64_t expiration_time = tablet_meta->file_cache_ttl_expiration_time(); if (!tablet->add_rowset_warmup_state(rs_meta, WarmUpTriggerSource::EVENT_DRIVEN)) { LOG(INFO) << "found duplicate warmup task for rowset " << rowset_id.to_string() diff --git a/be/src/cloud/cloud_rowset_builder.cpp b/be/src/cloud/cloud_rowset_builder.cpp index e03d347d94fe5e..29ece82d32088b 100644 --- a/be/src/cloud/cloud_rowset_builder.cpp +++ b/be/src/cloud/cloud_rowset_builder.cpp @@ -78,7 +78,7 @@ Status CloudRowsetBuilder::init() { context.mow_context = mow_context; context.write_file_cache = _req.write_file_cache; context.partial_update_info = _partial_update_info; - context.file_cache_ttl_sec = _tablet->ttl_seconds(); + context.file_cache_expiration_time = _tablet->file_cache_ttl_expiration_time(); context.storage_resource = _engine.get_storage_resource(_req.storage_vault_id); if (!context.storage_resource) { return Status::InternalError("vault id not found, maybe not sync, vault id {}", diff --git a/be/src/cloud/cloud_schema_change_job.cpp b/be/src/cloud/cloud_schema_change_job.cpp index 32b4386e0595e8..8066063dd282a9 100644 --- a/be/src/cloud/cloud_schema_change_job.cpp +++ b/be/src/cloud/cloud_schema_change_job.cpp @@ -391,7 +391,7 @@ Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParam // like the load and compaction output does. Otherwise it is cached in the // NORMAL/INDEX queues here, while every warm-up path downloads it into the TTL // queue on the destination cluster. - context.file_cache_ttl_sec = _new_tablet->ttl_seconds(); + context.file_cache_expiration_time = _new_tablet->file_cache_ttl_expiration_time(); context.tablet = _new_tablet; if (!context.storage_resource) { return Status::InternalError("vault id not found, maybe not sync, vault id {}", diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp index f2398128424ad5..b5c5f0425a0722 100644 --- a/be/src/cloud/cloud_tablet.cpp +++ b/be/src/cloud/cloud_tablet.cpp @@ -1908,11 +1908,7 @@ void CloudTablet::_add_rowsets_directly(std::vector& rowsets, continue; } - int64_t expiration_time = _tablet_meta->ttl_seconds() == 0 || - rowset_meta->newest_write_timestamp() <= 0 - ? 0 - : rowset_meta->newest_write_timestamp() + - _tablet_meta->ttl_seconds(); + int64_t expiration_time = _tablet_meta->file_cache_ttl_expiration_time(); g_file_cache_cloud_tablet_submitted_segment_num << 1; if (rs->rowset_meta()->segment_file_size(seg_id) > 0) { g_file_cache_cloud_tablet_submitted_segment_size diff --git a/be/src/cloud/cloud_warm_up_manager.cpp b/be/src/cloud/cloud_warm_up_manager.cpp index bc9b30b6273c88..c8b2a8a9eb95e9 100644 --- a/be/src/cloud/cloud_warm_up_manager.cpp +++ b/be/src/cloud/cloud_warm_up_manager.cpp @@ -302,7 +302,7 @@ void CloudWarmUpManager::handle_jobs() { continue; } - int64_t expiration_time = tablet_meta->ttl_seconds(); + int64_t expiration_time = tablet_meta->file_cache_ttl_expiration_time(); if (!tablet->add_rowset_warmup_state(*rs, WarmUpTriggerSource::JOB)) { LOG(INFO) << "found duplicate warmup task for rowset " << rs->rowset_id() << ", skip it"; diff --git a/be/src/exec/scan/olap_scanner.cpp b/be/src/exec/scan/olap_scanner.cpp index 04ede793ed579c..06e6a08c771e32 100644 --- a/be/src/exec/scan/olap_scanner.cpp +++ b/be/src/exec/scan/olap_scanner.cpp @@ -292,7 +292,7 @@ Status OlapScanner::prepare() { _tablet_reader_params.collection_statistics = std::make_shared(); auto io_ctx = build_score_runtime_collection_io_context( - _state, ReaderType::READER_QUERY, tablet->ttl_seconds(), + _state, ReaderType::READER_QUERY, tablet->file_cache_ttl_expiration_time(), &_tablet_reader->mutable_stats()->file_cache_stats); RETURN_IF_ERROR(_tablet_reader_params.collection_statistics->collect( diff --git a/be/src/io/fs/file_writer.h b/be/src/io/fs/file_writer.h index 9402fdef18303c..92e4475d63790d 100644 --- a/be/src/io/fs/file_writer.h +++ b/be/src/io/fs/file_writer.h @@ -47,7 +47,7 @@ struct FileWriterOptions { bool allow_adaptive_file_cache_write = true; bool is_cold_data = false; bool sync_file_data = true; // Whether flush data into storage system - uint64_t file_cache_expiration_time = 0; // Relative time + uint64_t file_cache_expiration_time = 0; // Absolute time, 0 means no TTL uint64_t approximate_bytes_to_write = 0; // Approximate bytes to write, used for file cache }; diff --git a/be/src/storage/compaction/compaction.cpp b/be/src/storage/compaction/compaction.cpp index a60ce91da5bbd2..449a49fbf64386 100644 --- a/be/src/storage/compaction/compaction.cpp +++ b/be/src/storage/compaction/compaction.cpp @@ -1867,7 +1867,7 @@ Status CloudCompactionMixin::construct_output_rowset_writer(RowsetWriterContext& // TODO(gavin): Ensure that the retention of hot data is implemented with precision. ctx.write_file_cache = should_cache_compaction_output(); - ctx.file_cache_ttl_sec = _tablet->ttl_seconds(); + ctx.file_cache_expiration_time = _tablet->file_cache_ttl_expiration_time(); ctx.approximate_bytes_to_write = _input_rowsets_total_size; // Set fine-grained control: only write index files to cache if configured diff --git a/be/src/storage/rowset/beta_rowset_reader.cpp b/be/src/storage/rowset/beta_rowset_reader.cpp index af3169eaf86973..a28572d54dc343 100644 --- a/be/src/storage/rowset/beta_rowset_reader.cpp +++ b/be/src/storage/rowset/beta_rowset_reader.cpp @@ -241,13 +241,7 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context _read_options.condition_cache_digest = _read_context->condition_cache_digest; } - _read_options.io_ctx.expiration_time = - read_context->ttl_seconds > 0 && _rowset->rowset_meta()->newest_write_timestamp() > 0 - ? _rowset->rowset_meta()->newest_write_timestamp() + read_context->ttl_seconds - : 0; - if (_read_options.io_ctx.expiration_time <= UnixSeconds()) { - _read_options.io_ctx.expiration_time = 0; - } + _read_options.io_ctx.expiration_time = read_context->file_cache_expiration_time; bool enable_segment_cache = true; auto* state = read_context->runtime_state; diff --git a/be/src/storage/rowset/rowset_reader_context.h b/be/src/storage/rowset/rowset_reader_context.h index 066c5890af005a..539a4576d650c9 100644 --- a/be/src/storage/rowset/rowset_reader_context.h +++ b/be/src/storage/rowset/rowset_reader_context.h @@ -88,7 +88,10 @@ struct RowsetReaderContext { RowsetId rowset_id; // slots that cast may be eliminated in storage layer std::map target_cast_type_for_variants; - int64_t ttl_seconds = 0; + // Absolute timestamp (seconds since epoch) after which cache blocks filled by this + // read stop being TTL protected; 0 means no TTL. + // See TabletMeta::file_cache_ttl_expiration_time(). + int64_t file_cache_expiration_time = 0; std::map virtual_column_exprs; std::map vir_cid_to_idx_in_block; diff --git a/be/src/storage/rowset/rowset_writer_context.h b/be/src/storage/rowset/rowset_writer_context.h index b65232f3c1c127..11ed54ac4339d9 100644 --- a/be/src/storage/rowset/rowset_writer_context.h +++ b/be/src/storage/rowset/rowset_writer_context.h @@ -104,7 +104,11 @@ struct RowsetWriterContext { /// begin file cache opts bool write_file_cache = false; bool is_hot_data = false; - uint64_t file_cache_ttl_sec = 0; + // Absolute timestamp (seconds since epoch) after which the cache blocks written by + // this rowset stop being TTL protected; 0 means no TTL. Always set it from + // BaseTablet::file_cache_ttl_expiration_time() so every writer agrees with the + // deadline BlockFileCacheTtlMgr sweeps by. + uint64_t file_cache_expiration_time = 0; uint64_t approximate_bytes_to_write = 0; // If true, compaction output only writes index files to file cache, not data files bool compaction_output_write_index_only = false; @@ -218,9 +222,7 @@ struct RowsetWriterContext { append_info.tablet_id = tablet_id; append_info.rowset_id = rowset_id.to_string(); append_info.txn_id = txn_id; - append_info.expiration_time = file_cache_ttl_sec > 0 && newest_write_timestamp > 0 - ? newest_write_timestamp + file_cache_ttl_sec - : 0; + append_info.expiration_time = file_cache_expiration_time; fs = std::make_shared(fs, append_info); } @@ -239,7 +241,7 @@ struct RowsetWriterContext { io::FileWriterOptions get_file_writer_options(FileType file_type = FileType::SEGMENT_FILE) { io::FileWriterOptions opts {.write_file_cache = write_file_cache, .is_cold_data = is_hot_data, - .file_cache_expiration_time = file_cache_ttl_sec, + .file_cache_expiration_time = file_cache_expiration_time, .approximate_bytes_to_write = approximate_bytes_to_write}; if (config::enable_file_cache_write_index_file_only) { diff --git a/be/src/storage/tablet/base_tablet.h b/be/src/storage/tablet/base_tablet.h index 73e797610b9d7e..9abad642144574 100644 --- a/be/src/storage/tablet/base_tablet.h +++ b/be/src/storage/tablet/base_tablet.h @@ -78,6 +78,10 @@ class BaseTablet : public std::enable_shared_from_this { KeysType keys_type() const { return _tablet_meta->tablet_schema()->keys_type(); } size_t num_key_columns() const { return _tablet_meta->tablet_schema()->num_key_columns(); } int64_t ttl_seconds() const { return _tablet_meta->ttl_seconds(); } + // See TabletMeta::file_cache_ttl_expiration_time(). + int64_t file_cache_ttl_expiration_time() const { + return _tablet_meta->file_cache_ttl_expiration_time(); + } // currently used by schema change, inverted index building, and cooldown std::timed_mutex& get_schema_change_lock() { return _schema_change_lock; } bool enable_unique_key_merge_on_write() const { diff --git a/be/src/storage/tablet/tablet_meta.cpp b/be/src/storage/tablet/tablet_meta.cpp index 345bf4546c8ad3..15abf78b549867 100644 --- a/be/src/storage/tablet/tablet_meta.cpp +++ b/be/src/storage/tablet/tablet_meta.cpp @@ -652,6 +652,25 @@ Status TabletMeta::save_meta(DataDir* data_dir) { return _save_meta(data_dir); } +int64_t TabletMeta::file_cache_ttl_expiration_time() const { + int64_t ttl = ttl_seconds(); + int64_t ctime = creation_time(); + if (ttl <= 0 || ctime <= 0) { + return 0; + } + // FE caps file_cache_ttl_seconds at Long.MAX_VALUE / 2, so this cannot wrap, but a tablet + // meta that reached us from anywhere else still must not turn a huge ttl into a past + // deadline that silently downgrades the tablet to normal cache. + if (ctime > std::numeric_limits::max() - ttl) { + return std::numeric_limits::max(); + } + int64_t expiration_time = ctime + ttl; + // Already past the deadline: report no TTL at all, so callers stamp the blocks they + // create as NORMAL right away instead of putting them in the TTL queue for + // BlockFileCacheTtlMgr to take straight back out again. + return expiration_time > UnixSeconds() ? expiration_time : 0; +} + Status TabletMeta::_save_meta(DataDir* data_dir) { // check if tablet uid is valid if (_tablet_uid.hi == 0 && _tablet_uid.lo == 0) { diff --git a/be/src/storage/tablet/tablet_meta.h b/be/src/storage/tablet/tablet_meta.h index 0f6d2bc69c3890..a9d06f8827abbd 100644 --- a/be/src/storage/tablet/tablet_meta.h +++ b/be/src/storage/tablet/tablet_meta.h @@ -317,6 +317,17 @@ class TabletMeta : public MetadataAdder { _ttl_seconds = ttl_seconds; } + // Absolute timestamp (seconds since epoch) at which this tablet's data stops being kept + // in the file cache TTL queue, or 0 when the tablet has no TTL or the deadline has + // already passed. The deadline is anchored at the tablet creation time, so every tablet + // of a table shares one deadline regardless of when each rowset was written. + // + // This is the single definition of that deadline. The load, compaction, schema change, + // query and warm up paths all stamp the cache blocks they create with this value, and + // BlockFileCacheTtlMgr expires those blocks by the very same value, so a block's + // recorded expiration time always agrees with the sweep that acts on it. + int64_t file_cache_ttl_expiration_time() const; + int64_t avg_rs_meta_serialize_size() const { return _avg_rs_meta_serialize_size; } EncryptionAlgorithmPB encryption_algorithm() const { return _encryption_algorithm; } diff --git a/be/src/storage/tablet/tablet_reader.cpp b/be/src/storage/tablet/tablet_reader.cpp index 7e3f3ffd701ab3..7e87419cc4d9f4 100644 --- a/be/src/storage/tablet/tablet_reader.cpp +++ b/be/src/storage/tablet/tablet_reader.cpp @@ -184,7 +184,7 @@ Status TabletReader::_capture_rs_readers(const ReaderParams& read_params) { _reader_context.common_expr_ctxs_push_down = read_params.common_expr_ctxs_push_down; _reader_context.output_columns = &read_params.output_columns; _reader_context.push_down_agg_type_opt = read_params.push_down_agg_type_opt; - _reader_context.ttl_seconds = _tablet->ttl_seconds(); + _reader_context.file_cache_expiration_time = _tablet->file_cache_ttl_expiration_time(); _reader_context.score_runtime = read_params.score_runtime; _reader_context.collection_statistics = read_params.collection_statistics; diff --git a/be/test/storage/tablet/tablet_meta_test.cpp b/be/test/storage/tablet/tablet_meta_test.cpp index 33b784f666f1e0..77e2626d7d71b3 100644 --- a/be/test/storage/tablet/tablet_meta_test.cpp +++ b/be/test/storage/tablet/tablet_meta_test.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include #include @@ -29,6 +30,7 @@ #include "storage/rowset/rowset.h" #include "storage/tablet/tablet_schema.h" #include "testutil/mock_rowset.h" +#include "util/time.h" namespace doris { @@ -403,4 +405,41 @@ TEST(TabletMetaTest, TestDeleteBitmap) { EXPECT_EQ(d.cardinality(), 500); } +TEST(TabletMetaTest, FileCacheTtlExpirationTime) { + TabletMeta meta; + + // No TTL configured: nothing to protect. + meta.set_creation_time(UnixSeconds() - 10); + meta.set_ttl_seconds(0); + EXPECT_EQ(0, meta.file_cache_ttl_expiration_time()); + + // Unknown creation time cannot anchor a deadline. + meta.set_creation_time(0); + meta.set_ttl_seconds(3600); + EXPECT_EQ(0, meta.file_cache_ttl_expiration_time()); + + // Live tablet: the deadline is creation time + ttl, an absolute timestamp. + int64_t ctime = UnixSeconds() - 10; + meta.set_creation_time(ctime); + meta.set_ttl_seconds(3600); + EXPECT_EQ(ctime + 3600, meta.file_cache_ttl_expiration_time()); + + // Past the deadline: report no TTL, so callers stamp new blocks as NORMAL instead of + // putting them in the TTL queue for the expiration sweep to take back out. + meta.set_creation_time(UnixSeconds() - 3600); + meta.set_ttl_seconds(60); + EXPECT_EQ(0, meta.file_cache_ttl_expiration_time()); + + // Exactly at the deadline counts as expired. + int64_t now = UnixSeconds(); + meta.set_creation_time(now - 60); + meta.set_ttl_seconds(60); + EXPECT_EQ(0, meta.file_cache_ttl_expiration_time()); + + // A ttl large enough to overflow must not wrap into a past deadline. + meta.set_creation_time(UnixSeconds()); + meta.set_ttl_seconds(std::numeric_limits::max()); + EXPECT_EQ(std::numeric_limits::max(), meta.file_cache_ttl_expiration_time()); +} + } // namespace doris diff --git a/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_seconds.groovy b/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_seconds.groovy index fd5d72fb842ae7..491576e56014f9 100644 --- a/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_seconds.groovy +++ b/regression-test/suites/cloud_p0/cache/ttl/alter_ttl_seconds.groovy @@ -139,7 +139,10 @@ suite("test_ttl_seconds") { load_customer_once("customer_ttl") def tabletIds = getTabletIds.call("customer_ttl") - waitForFileCacheType.call(tabletIds, "ttl", 15000L, 500L) + // No wait for the "ttl" type here. The TTL deadline is the tablet creation time plus + // file_cache_ttl_seconds, and with a 5s ttl the load itself outlives it, so most of the + // data is written straight into the normal queue and the table never has all of its + // blocks in the TTL queue at once. sleep(30000) // 30s getMetricsMethod.call() { respCode, body -> diff --git a/regression-test/suites/cloud_p0/cache/ttl/test_ttl_expired_tablet.groovy b/regression-test/suites/cloud_p0/cache/ttl/test_ttl_expired_tablet.groovy new file mode 100644 index 00000000000000..80b86d63f837a8 --- /dev/null +++ b/regression-test/suites/cloud_p0/cache/ttl/test_ttl_expired_tablet.groovy @@ -0,0 +1,207 @@ +// 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. + +import org.codehaus.groovy.runtime.IOGroovyMethods + +// The file cache TTL deadline is the tablet creation time plus file_cache_ttl_seconds. Once a +// tablet is past that deadline, the load and query paths must stamp the blocks they create as +// non-TTL right away, instead of creating TTL blocks for the background sweep to demote again. +// +// This test deliberately runs with the TTL background threads turned down to a 10 minute +// interval. That is what gives it teeth: the sweep would otherwise repair the cache type within +// a second or two and the assertions would hold no matter what the write and read paths did. +// With the sweep out of the way, the cache type observed here is purely the one chosen at +// admission, so the expired-tablet cases below fail if the deadline is not applied at the source. +suite("test_ttl_expired_tablet") { + def ttlSeconds = 30 + def sweepOffMs = 600000 + + def custoBeConfig = [ + enable_evict_file_cache_in_advance : false, + file_cache_enter_disk_resource_limit_mode_percent : 99, + // Long enough that neither TTL background thread reconciles anything while the test runs. + file_cache_background_ttl_gc_interval_ms : sweepOffMs, + file_cache_background_ttl_info_update_interval_ms : sweepOffMs, + file_cache_background_tablet_id_flush_interval_ms : 1000 + ] + + setBeConfigTemporary(custoBeConfig) { + sql "set global enable_auto_analyze = false" + sql "set global enable_audit_plugin = false" + def clusters = sql " SHOW CLUSTERS; " + assertTrue(!clusters.isEmpty()) + def validCluster = clusters[0][0] + sql """use @${validCluster};"""; + + String[][] backends = sql """ show backends """ + String backendId; + def backendIdToBackendIP = [:] + def backendIdToBackendHttpPort = [:] + for (String[] backend in backends) { + if (backend[9].equals("true") && backend[19].contains("${validCluster}")) { + backendIdToBackendIP.put(backend[0], backend[1]) + backendIdToBackendHttpPort.put(backend[0], backend[4]) + } + } + assertEquals(backendIdToBackendIP.size(), 1) + + backendId = backendIdToBackendIP.keySet()[0] + def url = backendIdToBackendIP.get(backendId) + ":" + backendIdToBackendHttpPort.get(backendId) + """/api/file_cache?op=clear&sync=true""" + def clearFileCache = { check_func -> + httpTest { + endpoint "" + uri url + op "get" + body "" + check check_func + } + } + + def getTabletIds = { String tableName -> + def tablets = sql "show tablets from ${tableName}" + assertTrue(tablets.size() > 0, "No tablets found for table ${tableName}") + tablets.collect { it[0] as Long } + } + + // Counts of each cache type across the given tablets, e.g. [normal: 40, index: 8]. + def cacheTypeCounts = { List tabletIds -> + def counts = [:].withDefault { 0 } + for (Long tabletId in tabletIds) { + def rows = sql "select type from information_schema.file_cache_info where tablet_id = ${tabletId}" + for (row in rows) { + counts[row[0].toString().toLowerCase()] += 1 + } + } + return counts + } + + def totalBlocks = { Map counts -> counts.values().sum() ?: 0 } + + // Cache writes are asynchronous, so wait for the blocks to show up before judging their + // type. Only their existence is waited on; the type assertions stay strict. + def waitForAnyBlock = { List tabletIds, long timeoutMs -> + long start = System.currentTimeMillis() + while (System.currentTimeMillis() - start < timeoutMs) { + if (totalBlocks(cacheTypeCounts.call(tabletIds)) > 0) { + return + } + sleep(1000) + } + assertTrue(false, "timed out waiting for cached blocks of tablets ${tabletIds}") + } + + def waitForEmptyCache = { List tabletIds, long timeoutMs -> + long start = System.currentTimeMillis() + while (System.currentTimeMillis() - start < timeoutMs) { + if (totalBlocks(cacheTypeCounts.call(tabletIds)) == 0) { + return + } + sleep(1000) + } + assertTrue(false, "timed out waiting for an empty cache for tablets ${tabletIds}") + } + + def loadCustomerRows = { String table -> + def totalRows = 200 + def batchSize = 100 + def commentSuffix = ' ' + ('X' * 50) + for (int offset = 0; offset < totalRows; offset += batchSize) { + def sb = new StringBuilder() + int batchEnd = Math.min(totalRows, offset + batchSize) + for (int idx = offset; idx < batchEnd; idx++) { + def customerId = 10001 + idx + def customerName = String.format('Customer#%09d', customerId) + sb.append("""INSERT INTO ${table} VALUES ( + ${customerId}, + '${customerName}', + 'Address Line 1', + 15, + '123-456-7890', + 12345.67, + 'AUTOMOBILE', + 'This is a test comment for the customer.${commentSuffix}' + ); + """) + } + sql sb.toString() + } + } + + def createCustomerTable = { String table, long ttl -> + def ddl = new File("""${context.file.parent}/../ddl/customer_ttl.sql""").text + sql (ddl.replace("customer_ttl", table) + + """ PROPERTIES("file_cache_ttl_seconds"="${ttl}") """) + sql """ alter table ${table} set ("disable_auto_compaction" = "true") """ + } + + sql """ DROP TABLE IF EXISTS customer_ttl_expired """ + sql """ DROP TABLE IF EXISTS customer_ttl_live """ + clearFileCache.call() { respCode, body -> {} } + sleep(5000) + + // --------------------------------------------------------------------------------------- + // Case 1: load into a tablet that is already past its deadline. + // --------------------------------------------------------------------------------------- + createCustomerTable.call("customer_ttl_expired", ttlSeconds) + // Let the deadline pass before a single row is written. + sleep((ttlSeconds + 15) * 1000L) + + loadCustomerRows("customer_ttl_expired") + def expiredTablets = getTabletIds.call("customer_ttl_expired") + + // Guard against a vacuous pass: the load must actually have cached something. + waitForAnyBlock.call(expiredTablets, 60000L) + def afterLoad = cacheTypeCounts.call(expiredTablets) + logger.info("cache types after loading an expired tablet: ${afterLoad}") + assertEquals(0, afterLoad['ttl'], + "load path put blocks of an expired tablet into the TTL queue: ${afterLoad}") + + // --------------------------------------------------------------------------------------- + // Case 2: read from that tablet on a cold cache, so the query path admits the blocks. + // --------------------------------------------------------------------------------------- + clearFileCache.call() { respCode, body -> {} } + // Without an empty cache the query below would be a hit and would admit nothing, leaving + // the read path untested. + waitForEmptyCache.call(expiredTablets, 60000L) + + // sum() has to read the column data, unlike count(*) which can be answered from metadata. + sql """ select sum(C_ACCTBAL), count(C_COMMENT) from customer_ttl_expired """ + + waitForAnyBlock.call(expiredTablets, 60000L) + def afterRead = cacheTypeCounts.call(expiredTablets) + logger.info("cache types after reading an expired tablet: ${afterRead}") + assertEquals(0, afterRead['ttl'], + "read path put blocks of an expired tablet into the TTL queue: ${afterRead}") + + // --------------------------------------------------------------------------------------- + // Case 3: a tablet still inside its window must keep using the TTL queue. Without this the + // two cases above would also pass if the deadline were simply always reported as expired. + // --------------------------------------------------------------------------------------- + createCustomerTable.call("customer_ttl_live", 3600) + loadCustomerRows("customer_ttl_live") + def liveTablets = getTabletIds.call("customer_ttl_live") + + waitForAnyBlock.call(liveTablets, 60000L) + def liveCounts = cacheTypeCounts.call(liveTablets) + logger.info("cache types for a tablet inside its TTL window: ${liveCounts}") + assertTrue(liveCounts['ttl'] > 0, + "tablet inside its TTL window got no TTL blocks: ${liveCounts}") + + sql """ DROP TABLE IF EXISTS customer_ttl_expired """ + sql """ DROP TABLE IF EXISTS customer_ttl_live """ + } +}