[test](cloud) Time travel: unit tests, regression tests, and design doc

Author: Sivarajan Narayanan

be/src/cloud/cloud_meta_mgr.cpp

@@ -842,10 +842,9 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
                 // 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);
+                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)
@@ -857,9 +856,8 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
                     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(_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));
             }

be/src/cloud/cloud_tablet.cpp

@@ -523,10 +523,8 @@ void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
 }
 
 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) {
+        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);
@@ -538,16 +536,15 @@ Status CloudTablet::capture_rs_readers_with_tt_rowsets(
     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();
+            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();
-              });
+    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) {

be/src/cloud/cloud_tablet.h

@@ -160,11 +160,10 @@ class CloudTablet final : public BaseTablet {
 
     // 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);
+    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,

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,10 +48,9 @@
 #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 "cloud/pb_convert.h"
-#include "storage/rowset/rowset_factory.h"
 #include "storage/tablet/tablet_manager.h"
 #include "util/to_string.h"
 
@@ -844,19 +844,17 @@ Status OlapScanLocalState::_sync_cloud_tablets(RuntimeState* state) {
             // 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;
+            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(),
+                                _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();
@@ -881,8 +879,8 @@ Status OlapScanLocalState::_sync_cloud_tablets(RuntimeState* state) {
                         }
                     });
                     // 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));
+                    auto tablet =
+                            DORIS_TRY(ExecEnv::get_tablet(_scan_ranges[i]->tablet_id, sync_stats));
                     _tablets[i] = {std::move(tablet), version, {}};
                     SyncOptions options;
                     options.query_version = version;
@@ -911,8 +909,7 @@ Status OlapScanLocalState::_sync_cloud_tablets(RuntimeState* state) {
                                 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()) {
+                                if (RowsetFactory::create_rowset(schema, "", rs_meta, &rs).ok()) {
                                     _tablets[i].tt_extra_rowsets.push_back(std::move(rs));
                                 }
                             }
@@ -1030,12 +1027,10 @@ Status OlapScanLocalState::prepare(RuntimeState* state) {
         // 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);
+            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()));
+                    &_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},
@@ -1044,14 +1039,12 @@ Status OlapScanLocalState::prepare(RuntimeState* state) {
                              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,
+                     .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}));
+                             config::is_cloud_mode() ? _state->query_freshness_tolerance_ms()
+                                                     : -1}));
         }
         if (!PipelineXLocalState<>::_state->skip_delete_predicate()) {
             _read_sources[i].fill_delete_predicates();

cloud/src/meta-service/meta_service.cpp

@@ -405,7 +405,7 @@ static int64_t resolve_version_at_time(Transaction* txn, const std::string& inst
     FullRangeGetOptions options;
     options.reverse = true;
     options.begin_key_selector = RangeKeySelector::FIRST_GREATER_OR_EQUAL;
-    options.end_key_selector   = RangeKeySelector::FIRST_GREATER_THAN;
+    options.end_key_selector = RangeKeySelector::FIRST_GREATER_THAN;
 
     std::unique_ptr<FullRangeGetIterator> iter =
             txn->full_range_get(begin_key, end_key, std::move(options));
@@ -414,7 +414,7 @@ static int64_t resolve_version_at_time(Transaction* txn, const std::string& inst
         auto item = iter->next();
         if (!item.has_value()) break;
 
-        std::string_view raw_key   = item->first;
+        std::string_view raw_key = item->first;
         std::string_view raw_value = item->second;
 
         // Strip the trailing versionstamp suffix before deserialising the value.
@@ -464,7 +464,7 @@ void MetaServiceImpl::get_version_at_time(::google::protobuf::RpcController* con
         return;
     }
 
-    int64_t timestamp_ms   = request->timestamp_ms();
+    int64_t timestamp_ms = request->timestamp_ms();
     int32_t retention_days = request->has_retention_days() ? request->retention_days() : 90;
 
     // Validate timestamp window — same logic for both single and batch.
@@ -482,8 +482,8 @@ void MetaServiceImpl::get_version_at_time(::google::protobuf::RpcController* con
     // Allow configurable clock skew to prevent spurious future-timestamp rejections.
     if (timestamp_ms > now_ms + config::time_travel_clock_skew_tolerance_ms) {
         code = MetaServiceCode::INVALID_ARGUMENT;
-        msg = fmt::format("Requested timestamp {} ms is in the future (now={} ms).",
-                          timestamp_ms, now_ms);
+        msg = fmt::format("Requested timestamp {} ms is in the future (now={} ms).", timestamp_ms,
+                          now_ms);
         return;
     }
 
@@ -510,9 +510,9 @@ void MetaServiceImpl::get_version_at_time(::google::protobuf::RpcController* con
         std::vector<int64_t> resolved_times(n);
         for (int i = 0; i < n; ++i) {
             int64_t update_time_ms = -1;
-            int64_t version = resolve_version_at_time(txn.get(), instance_id,
-                                                      request->partition_ids(i),
-                                                      timestamp_ms, &update_time_ms);
+            int64_t version =
+                    resolve_version_at_time(txn.get(), instance_id, request->partition_ids(i),
+                                            timestamp_ms, &update_time_ms);
             if (version == -2) {
                 code = MetaServiceCode::KV_TXN_GET_ERR;
                 msg = fmt::format("FDB scan error during time travel for partition {}",
@@ -543,30 +543,29 @@ void MetaServiceImpl::get_version_at_time(::google::protobuf::RpcController* con
                     // Each encoded int64 in the key is 9 bytes (1 tag + 8 data).
                     std::string ck_begin, ck_end;
                     tt_compaction_key({instance_id, tablet_id, 0, 0}, &ck_begin);
-                    tt_compaction_key({instance_id, tablet_id,
-                                       std::numeric_limits<int64_t>::max(),
-                                       std::numeric_limits<int64_t>::max()}, &ck_end);
+                    tt_compaction_key({instance_id, tablet_id, std::numeric_limits<int64_t>::max(),
+                                       std::numeric_limits<int64_t>::max()},
+                                      &ck_end);
                     ck_end.push_back('\xff');
 
                     std::unique_ptr<RangeGetIterator> ck_iter;
                     bool found_checkpoint = false;
-                    if (txn->get(ck_begin, ck_end, &ck_iter) == TxnErrorCode::TXN_OK
-                            && ck_iter) {
+                    if (txn->get(ck_begin, ck_end, &ck_iter) == TxnErrorCode::TXN_OK && ck_iter) {
                         // Trailing key layout: start_v(9) + end_v(9) = 18 bytes.
                         int64_t best_start = INT64_MAX;
                         std::string best_val;
                         while (ck_iter->has_next()) {
                             auto [k, v] = ck_iter->next();
                             if (k.size() < 18) continue;
                             std::string_view sv = k.substr(k.size() - 18, 9);
-                            std::string_view ev = k.substr(k.size() - 9,  9);
+                            std::string_view ev = k.substr(k.size() - 9, 9);
                             int64_t start_v = 0, end_v = 0;
                             if (decode_int64(&sv, &start_v) != 0) continue;
-                            if (decode_int64(&ev, &end_v)   != 0) continue;
+                            if (decode_int64(&ev, &end_v) != 0) continue;
                             if (start_v <= version && version <= end_v) {
                                 if (start_v < best_start) {
                                     best_start = start_v;
-                                    best_val   = std::string(v);
+                                    best_val = std::string(v);
                                 }
                                 found_checkpoint = true;
                             }
@@ -605,8 +604,7 @@ void MetaServiceImpl::get_version_at_time(::google::protobuf::RpcController* con
     }
 
     int64_t update_time_ms = -1;
-    int64_t version = resolve_version_at_time(txn.get(), instance_id,
-                                              request->partition_id(),
+    int64_t version = resolve_version_at_time(txn.get(), instance_id, request->partition_id(),
                                               timestamp_ms, &update_time_ms);
     if (version == -2) {
         code = MetaServiceCode::KV_TXN_GET_ERR;
@@ -626,10 +624,10 @@ void MetaServiceImpl::get_version_at_time(::google::protobuf::RpcController* con
     response->set_version_update_time_ms(update_time_ms);
 }
 
-void MetaServiceImpl::disable_time_travel_table(
-        ::google::protobuf::RpcController* controller,
-        const DisableTimeTravelTableRequest* request, DisableTimeTravelTableResponse* response,
-        ::google::protobuf::Closure* done) {
+void MetaServiceImpl::disable_time_travel_table(::google::protobuf::RpcController* controller,
+                                                const DisableTimeTravelTableRequest* request,
+                                                DisableTimeTravelTableResponse* response,
+                                                ::google::protobuf::Closure* done) {
     RPC_PREPROCESS(disable_time_travel_table, get, del);
 
     if (!request->has_table_id() || request->table_id() <= 0) {
@@ -656,8 +654,7 @@ void MetaServiceImpl::disable_time_travel_table(
     err = txn->commit();
     if (err != TxnErrorCode::TXN_OK) {
         code = cast_as<ErrCategory::COMMIT>(err);
-        msg = fmt::format("failed to remove time travel marker, table_id={} err={}", table_id,
-                          err);
+        msg = fmt::format("failed to remove time travel marker, table_id={} err={}", table_id, err);
         return;
     }