[Feat](TimeTravel) Time Travel for Cloud / Decoupled Mode

be/src/cloud/cloud_meta_mgr.cpp

@@ -810,24 +810,57 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
             DeleteBitmap delete_bitmap(tablet_id);
             int64_t old_max_version = req.start_version() - 1;
             auto read_version = config::delete_bitmap_store_read_version;
-            auto st = sync_tablet_delete_bitmap(tablet, old_max_version, resp.rowset_meta(),
-                                                resp.stats(), req.idx(), &delete_bitmap,
-                                                options.full_sync, sync_stats, read_version, false);
-            if (st.is<ErrorCode::ROWSETS_EXPIRED>() && tried++ < retry_times) {
-                LOG_INFO("rowset meta is expired, need to retry, " + tablet_info)
-                        .tag("tried", tried)
-                        .error(st);
-                continue;
-            }
-            if (!st.ok()) {
-                LOG_WARNING("failed to get delete bitmap, " + tablet_info).error(st);
-                return st;
+
+            // Time travel: filter rowset metas to only those visible at the historical version.
+            // Materialize into a vector because sync_tablet_delete_bitmap calls rbegin()
+            // (requiring a bidirectional range) and because RepeatedPtrField with filter_view
+            // is a forward-only range.
+            if (options.delete_bitmap_max_version > 0) {
+                google::protobuf::RepeatedPtrField<RowsetMetaCloudPB> filtered_metas;
+                for (const auto& rm : resp.rowset_meta()) {
+                    if (rm.end_version() <= options.delete_bitmap_max_version) {
+                        *filtered_metas.Add() = rm;
+                    }
+                }
+                if (!filtered_metas.empty()) {
+                    auto st = sync_tablet_delete_bitmap(
+                            tablet, old_max_version, filtered_metas, resp.stats(), req.idx(),
+                            &delete_bitmap, options.full_sync, sync_stats, read_version, false);
+                    if (st.is<ErrorCode::ROWSETS_EXPIRED>() && tried++ < retry_times) {
+                        LOG_INFO("rowset meta is expired during time travel, retrying")
+                                .tag("tried", tried)
+                                .error(st);
+                        continue;
+                    }
+                    if (!st.ok()) {
+                        LOG_WARNING("failed to get delete bitmap for time travel").error(st);
+                        return st;
+                    }
+                }
+                // _log_mow_delete_bitmap and _check_delete_bitmap_v2_correctness compare
+                // the bitmap against ALL current rowsets and would produce false failures on
+                // the historical subset — skip them for the time travel path.
+                tablet->tablet_meta()->delete_bitmap().merge(delete_bitmap);
+            } else {
+                auto st = sync_tablet_delete_bitmap(
+                        tablet, old_max_version, resp.rowset_meta(), resp.stats(), req.idx(),
+                        &delete_bitmap, options.full_sync, sync_stats, read_version, false);
+                if (st.is<ErrorCode::ROWSETS_EXPIRED>() && tried++ < retry_times) {
+                    LOG_INFO("rowset meta is expired, need to retry, " + tablet_info)
+                            .tag("tried", tried)
+                            .error(st);
+                    continue;
+                }
+                if (!st.ok()) {
+                    LOG_WARNING("failed to get delete bitmap, " + tablet_info).error(st);
+                    return st;
+                }
+                tablet->tablet_meta()->delete_bitmap().merge(delete_bitmap);
+                RETURN_IF_ERROR(_log_mow_delete_bitmap(tablet, resp, delete_bitmap, old_max_version,
+                                                       options.full_sync, read_version));
+                RETURN_IF_ERROR(
+                        _check_delete_bitmap_v2_correctness(tablet, req, resp, old_max_version));
             }
-            tablet->tablet_meta()->delete_bitmap().merge(delete_bitmap);
-            RETURN_IF_ERROR(_log_mow_delete_bitmap(tablet, resp, delete_bitmap, old_max_version,
-                                                   options.full_sync, read_version));
-            RETURN_IF_ERROR(
-                    _check_delete_bitmap_v2_correctness(tablet, req, resp, old_max_version));
         }
         DBUG_EXECUTE_IF("CloudMetaMgr::sync_tablet_rowsets.before.modify_tablet_meta", {
             auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);

be/src/cloud/cloud_tablet.cpp

@@ -20,6 +20,7 @@
 #include <bvar/bvar.h>
 #include <bvar/latency_recorder.h>
 #include <gen_cpp/Types_types.h>
+#include <gen_cpp/cloud.pb.h>
 #include <gen_cpp/olap_file.pb.h>
 #include <rapidjson/document.h>
 #include <rapidjson/encodings.h>
@@ -43,6 +44,7 @@
 #include "cloud/cloud_tablet_mgr.h"
 #include "cloud/cloud_warm_up_manager.h"
 #include "cloud/config.h"
+#include "cloud/pb_convert.h"
 #include "common/cast_set.h"
 #include "common/config.h"
 #include "common/logging.h"
@@ -520,6 +522,52 @@ void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
     _tablet_meta->modify_rs_metas({}, rs_metas, false);
 }
 
+Status CloudTablet::capture_rs_readers_with_tt_rowsets(
+        const Version& spec_version, const std::vector<RowsetSharedPtr>& tt_extra_rowsets,
+        std::vector<RowSetSplits>* rs_splits, bool skip_missing_version) {
+    // Build a version path from the shared rowset map, then merge in tt_extra_rowsets
+    // for versions already compacted away. Only a shared lock is held; tablet state is unchanged.
+    std::shared_lock rlock(_meta_lock);
+
+    std::vector<RowsetSharedPtr> path;
+    for (auto& [ver, rs] : _rs_version_map) {
+        if (ver.second <= spec_version.second) path.push_back(rs);
+    }
+    for (auto& rs : tt_extra_rowsets) {
+        if (!rs || rs->end_version() > spec_version.second) continue;
+        bool covered = std::any_of(path.begin(), path.end(), [&rs](const RowsetSharedPtr& a) {
+            return a->start_version() <= rs->start_version() &&
+                   rs->end_version() <= a->end_version();
+        });
+        if (!covered) path.push_back(rs);
+    }
+
+    std::sort(path.begin(), path.end(), [](const RowsetSharedPtr& a, const RowsetSharedPtr& b) {
+        return a->start_version() < b->start_version();
+    });
+
+    int64_t expected = 0;
+    for (auto& rs : path) {
+        if (rs->start_version() != expected) {
+            if (!skip_missing_version) {
+                return Status::Error<ErrorCode::CAPTURE_ROWSET_READER_ERROR>(
+                        "missed versions: expected {} got {} tablet={}", expected,
+                        rs->start_version(), tablet_id());
+            }
+            break;
+        }
+        expected = rs->end_version() + 1;
+    }
+
+    rs_splits->clear();
+    for (auto& rs : path) {
+        RowsetReaderSharedPtr reader;
+        RETURN_IF_ERROR(rs->create_reader(&reader));
+        rs_splits->push_back({std::move(reader)});
+    }
+    return Status::OK();
+}
+
 void CloudTablet::delete_rowsets_for_schema_change(const std::vector<RowsetSharedPtr>& to_delete,
                                                    std::unique_lock<BthreadSharedMutex>&,
                                                    bool recycle_deleted_rowsets) {

be/src/cloud/cloud_tablet.h

@@ -66,6 +66,9 @@ struct SyncOptions {
     bool full_sync = false;
     bool merge_schema = false;
     int64_t query_version = -1;
+    // Time travel: if > 0, cap the delete bitmap fetch to this version so MoW
+    // tables see the correct historical delete state instead of current state.
+    int64_t delete_bitmap_max_version = -1;
 };
 
 struct RecycledRowsets {
@@ -155,6 +158,13 @@ class CloudTablet final : public BaseTablet {
                      std::unique_lock<BthreadSharedMutex>& meta_lock,
                      bool warmup_delta_data = false);
 
+    // Build a read source for spec_version, merging tt_extra_rowsets for compacted-away versions.
+    // Does not modify shared tablet state; caller provides pre-compaction rowsets from FDB.
+    Status capture_rs_readers_with_tt_rowsets(const Version& spec_version,
+                                              const std::vector<RowsetSharedPtr>& tt_extra_rowsets,
+                                              std::vector<RowSetSplits>* rs_splits,
+                                              bool skip_missing_version = false);
+
     // MUST hold EXCLUSIVE `_meta_lock`.
     void delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
                         std::unique_lock<BthreadSharedMutex>& meta_lock);

be/src/cloud/pb_convert.cpp

@@ -778,6 +778,7 @@ void doris_tablet_meta_to_cloud(TabletMetaCloudPB* out, const TabletMetaPB& in)
     out->set_is_persistent(in.is_persistent());
     out->set_table_name(in.table_name());
     out->set_ttl_seconds(in.ttl_seconds());
+    out->set_time_travel_retention_days(in.time_travel_retention_days());
     if (in.has_schema_version()) {
         out->set_schema_version(in.schema_version());
     }
@@ -872,6 +873,7 @@ void doris_tablet_meta_to_cloud(TabletMetaCloudPB* out, TabletMetaPB&& in) {
     out->set_is_persistent(in.is_persistent());
     out->set_table_name(in.table_name());
     out->set_ttl_seconds(in.ttl_seconds());
+    out->set_time_travel_retention_days(in.time_travel_retention_days());
     if (in.has_schema_version()) {
         out->set_schema_version(in.schema_version());
     }
@@ -970,6 +972,7 @@ void cloud_tablet_meta_to_doris(TabletMetaPB* out, const TabletMetaCloudPB& in)
     out->set_is_persistent(in.is_persistent());
     out->set_table_name(in.table_name());
     out->set_ttl_seconds(in.ttl_seconds());
+    out->set_time_travel_retention_days(in.time_travel_retention_days());
     if (in.has_schema_version()) {
         out->set_schema_version(in.schema_version());
     }
@@ -1064,6 +1067,7 @@ void cloud_tablet_meta_to_doris(TabletMetaPB* out, TabletMetaCloudPB&& in) {
     out->set_is_persistent(in.is_persistent());
     out->set_table_name(in.table_name());
     out->set_ttl_seconds(in.ttl_seconds());
+    out->set_time_travel_retention_days(in.time_travel_retention_days());
     if (in.has_schema_version()) {
         out->set_schema_version(in.schema_version());
     }

be/src/exec/operator/olap_scan_operator.cpp

@@ -29,6 +29,7 @@
 #include "cloud/cloud_tablet.h"
 #include "cloud/cloud_tablet_hotspot.h"
 #include "cloud/config.h"
+#include "cloud/pb_convert.h"
 #include "exec/operator/scan_operator.h"
 #include "exec/runtime_filter/runtime_filter_consumer_helper.h"
 #include "exec/scan/olap_scanner.h"
@@ -47,6 +48,7 @@
 #include "runtime/runtime_state.h"
 #include "service/backend_options.h"
 #include "storage/index/ann/ann_topn_runtime.h"
+#include "storage/rowset/rowset_factory.h"
 #include "storage/storage_engine.h"
 #include "storage/tablet/tablet.h"
 #include "storage/tablet/tablet_manager.h"
@@ -838,15 +840,26 @@ Status OlapScanLocalState::_sync_cloud_tablets(RuntimeState* state) {
             _tablets.resize(_scan_ranges.size());
             std::vector<std::function<Status()>> tasks;
             _sync_statistics.resize(_scan_ranges.size());
+
+            // FE has already resolved timestamp_ms → partition version and stored it in
+            // TPaloScanRange.version. tt_timestamp_ms is retained here for MoW delete bitmap capping.
+            const auto& p = _parent->cast<OlapScanOperatorX>();
+            const int64_t tt_timestamp_ms = p._olap_scan_node.__isset.time_travel_timestamp_ms
+                                                    ? p._olap_scan_node.time_travel_timestamp_ms
+                                                    : -1;
+
             for (size_t i = 0; i < _scan_ranges.size(); i++) {
                 auto* sync_stats = &_sync_statistics[i];
+                // Version is read from scan range for both normal and time-travel queries.
+                // For time travel, FE pre-resolved it via batch getVersionsAtTime().
                 int64_t version = 0;
                 std::from_chars(_scan_ranges[i]->version.data(),
                                 _scan_ranges[i]->version.data() + _scan_ranges[i]->version.size(),
                                 version);
                 auto task_ctx = state->get_task_execution_context();
                 auto task_create_time = std::chrono::steady_clock::now();
-                tasks.emplace_back([this, sync_stats, version, i, task_ctx, task_create_time]() {
+                tasks.emplace_back([this, sync_stats, version, i, task_ctx, task_create_time,
+                                    tt_timestamp_ms, &p]() mutable {
                     // Record bthread scheduling delay
                     auto task_start_time = std::chrono::steady_clock::now();
                     if (sync_stats) {
@@ -865,14 +878,43 @@ Status OlapScanLocalState::_sync_cloud_tablets(RuntimeState* state) {
                             _sync_cloud_tablets_watcher.stop();
                         }
                     });
+                    // Fetch tablet. Version is already resolved by FE — use it directly.
                     auto tablet =
                             DORIS_TRY(ExecEnv::get_tablet(_scan_ranges[i]->tablet_id, sync_stats));
-                    _tablets[i] = {std::move(tablet), version};
+                    _tablets[i] = {std::move(tablet), version, {}};
                     SyncOptions options;
                     options.query_version = version;
                     options.merge_schema = true;
+                    if (tt_timestamp_ms > 0) {
+                        // MoW: cap delete bitmap fetch to the historical version so the
+                        // tablet sees the correct delete state at that point in time.
+                        options.delete_bitmap_max_version = version;
+                    }
                     RETURN_IF_ERROR(std::dynamic_pointer_cast<CloudTablet>(_tablets[i].tablet)
                                             ->sync_rowsets(options, sync_stats));
+
+                    // Build scan-local extra rowsets from manifests for post-compaction TT reads.
+                    // These rowsets are never injected into the shared tablet state.
+                    if (p._olap_scan_node.__isset.tt_rowset_manifests) {
+                        const auto& manifests = p._olap_scan_node.tt_rowset_manifests;
+                        auto it = manifests.find(_scan_ranges[i]->tablet_id);
+                        if (it != manifests.end()) {
+                            _tablets[i].tt_has_manifest = true;
+                            for (const auto& bytes : it->second) {
+                                doris::RowsetMetaCloudPB cloud_meta;
+                                if (!cloud_meta.ParseFromString(bytes)) continue;
+                                RowsetMetaPB rs_meta_pb =
+                                        cloud::cloud_rowset_meta_to_doris(std::move(cloud_meta));
+                                auto rs_meta = std::make_shared<RowsetMeta>();
+                                rs_meta->init_from_pb(rs_meta_pb);
+                                RowsetSharedPtr rs;
+                                auto schema = _tablets[i].tablet->tablet_schema();
+                                if (RowsetFactory::create_rowset(schema, "", rs_meta, &rs).ok()) {
+                                    _tablets[i].tt_extra_rowsets.push_back(std::move(rs));
+                                }
+                            }
+                        }
+                    }
                     // FIXME(plat1ko): Avoid pointer cast
                     ExecEnv::GetInstance()->storage_engine().to_cloud().tablet_hotspot().count(
                             *_tablets[i].tablet);
@@ -977,21 +1019,33 @@ Status OlapScanLocalState::prepare(RuntimeState* state) {
                             _scan_ranges[i]->version.data() + _scan_ranges[i]->version.size(),
                             version);
             auto tablet = DORIS_TRY(ExecEnv::get_tablet(_scan_ranges[i]->tablet_id));
-            _tablets[i] = {std::move(tablet), version};
+            _tablets[i] = {std::move(tablet), version, {}};
         }
     }
 
     for (size_t i = 0; i < _scan_ranges.size(); i++) {
-        _read_sources[i] = DORIS_TRY(_tablets[i].tablet->capture_read_source(
-                {0, _tablets[i].version},
-                {.skip_missing_versions = _state->skip_missing_version(),
-                 .enable_fetch_rowsets_from_peers = config::enable_fetch_rowsets_from_peer_replicas,
-                 .capture_row_binlog = olap_scan_node().__isset.read_row_binlog &&
-                                       olap_scan_node().read_row_binlog,
-                 .enable_prefer_cached_rowset =
-                         config::is_cloud_mode() ? _state->enable_prefer_cached_rowset() : false,
-                 .query_freshness_tolerance_ms =
-                         config::is_cloud_mode() ? _state->query_freshness_tolerance_ms() : -1}));
+        // For time-travel reads with a manifest, augment the version path with
+        // pre-compaction rowsets without modifying the shared tablet state.
+        if (!_tablets[i].tt_extra_rowsets.empty() || _tablets[i].tt_has_manifest) {
+            auto cloud_tablet = std::dynamic_pointer_cast<CloudTablet>(_tablets[i].tablet);
+            RETURN_IF_ERROR(cloud_tablet->capture_rs_readers_with_tt_rowsets(
+                    {0, _tablets[i].version}, _tablets[i].tt_extra_rowsets,
+                    &_read_sources[i].rs_splits, _state->skip_missing_version()));
+        } else {
+            _read_sources[i] = DORIS_TRY(_tablets[i].tablet->capture_read_source(
+                    {0, _tablets[i].version},
+                    {.skip_missing_versions = _state->skip_missing_version(),
+                     .enable_fetch_rowsets_from_peers =
+                             config::enable_fetch_rowsets_from_peer_replicas,
+                     .capture_row_binlog = olap_scan_node().__isset.read_row_binlog &&
+                                           olap_scan_node().read_row_binlog,
+                     .enable_prefer_cached_rowset = config::is_cloud_mode()
+                                                            ? _state->enable_prefer_cached_rowset()
+                                                            : false,
+                     .query_freshness_tolerance_ms =
+                             config::is_cloud_mode() ? _state->query_freshness_tolerance_ms()
+                                                     : -1}));
+        }
         if (!PipelineXLocalState<>::_state->skip_delete_predicate()) {
             _read_sources[i].fill_delete_predicates();
         }

be/src/exec/scan/parallel_scanner_builder.cpp

@@ -62,7 +62,7 @@ Status ParallelScannerBuilder::build_scanners(std::list<ScannerSPtr>& scanners)
 Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>& scanners) {
     DCHECK_GE(_rows_per_scanner, _min_rows_per_scanner);
 
-    for (auto&& [tablet, version] : _tablets) {
+    for (auto&& [tablet, version, tt_extra_rowsets, tt_has_manifest] : _tablets) {
         DCHECK(_all_read_sources.contains(tablet->tablet_id()));
         auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
 
@@ -190,7 +190,7 @@ Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
 Status ParallelScannerBuilder::_build_scanners_by_per_segment(std::list<ScannerSPtr>& scanners) {
     DCHECK_GE(_rows_per_scanner, _min_rows_per_scanner);
 
-    for (auto&& [tablet, version] : _tablets) {
+    for (auto&& [tablet, version, tt_extra_rowsets, tt_has_manifest] : _tablets) {
         DCHECK(_all_read_sources.contains(tablet->tablet_id()));
         auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
 
@@ -244,7 +244,7 @@ Status ParallelScannerBuilder::_load() {
     bool enable_segment_cache = _state->query_options().__isset.enable_segment_cache
                                         ? _state->query_options().enable_segment_cache
                                         : true;
-    for (auto&& [tablet, version] : _tablets) {
+    for (auto&& [tablet, version, tt_extra_rowsets, tt_has_manifest] : _tablets) {
         const auto tablet_id = tablet->tablet_id();
         _all_read_sources[tablet_id] = _read_sources[idx];
         const auto& read_source = _all_read_sources[tablet_id];