branch-4.1: [fix](filecache) Unify the file cache TTL deadline across all paths

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()

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 {}",

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 {}",

be/src/cloud/cloud_tablet.cpp

@@ -1908,11 +1908,7 @@ void CloudTablet::_add_rowsets_directly(std::vector<RowsetSharedPtr>& 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

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";

be/src/exec/scan/olap_scanner.cpp

@@ -292,7 +292,7 @@ Status OlapScanner::prepare() {
         _tablet_reader_params.collection_statistics = std::make_shared<CollectionStatistics>();
 
         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(

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
 };
 

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

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;

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<std::string, DataTypePtr> 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<ColumnId, VExprContextSPtr> virtual_column_exprs;
     std::map<ColumnId, size_t> vir_cid_to_idx_in_block;

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<io::PackedFileSystem>(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) {

be/src/storage/tablet/base_tablet.h

@@ -78,6 +78,10 @@ class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
     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 {

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<int64_t>::max() - ttl) {
+        return std::numeric_limits<int64_t>::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) {

be/src/storage/tablet/tablet_meta.h

@@ -317,6 +317,17 @@ class TabletMeta : public MetadataAdder<TabletMeta> {
         _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; }

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;
 

be/test/storage/tablet/tablet_meta_test.cpp

@@ -20,6 +20,7 @@
 #include <gtest/gtest-message.h>
 #include <gtest/gtest-test-part.h>
 
+#include <limits>
 #include <memory>
 #include <string>
 #include <utility>
@@ -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<int64_t>::max());
+    EXPECT_EQ(std::numeric_limits<int64_t>::max(), meta.file_cache_ttl_expiration_time());
+}
+
 } // namespace doris

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 ->