resolved_ts: pass full safe_ts timestamp to observer
close tikv/tikv#13896
@@ -35,7 +35,7 @@ use tikv_util::{
Either,
};
use time::{Duration, Timespec};
-use txn_types::{TimeStamp, WriteBatchFlags};
+use txn_types::WriteBatchFlags;
use super::{metrics::PEER_ADMIN_CMD_COUNTER_VEC, peer_storage, Config};
use crate::{coprocessor::CoprocessorHost, store::snap::SNAPSHOT_VERSION, Error, Result};
@@ -1186,11 +1186,7 @@ impl RegionReadProgress {
if !core.pause {
self.safe_ts.store(ts, AtomicOrdering::Release);
// No need to update leader safe ts here.
- coprocessor.on_update_safe_ts(
- core.region_id,
- TimeStamp::new(ts).physical(),
- INVALID_TIMESTAMP,
- )
+ coprocessor.on_update_safe_ts(core.region_id, ts, INVALID_TIMESTAMP)
}
}
}
@@ -1232,11 +1228,7 @@ impl RegionReadProgress {
self.safe_ts.store(ts, AtomicOrdering::Release);
// After region merge, self safe ts may decrease, so leader safe ts should be
// reset.
- coprocessor.on_update_safe_ts(
- core.region_id,
- TimeStamp::new(ts).physical(),
- TimeStamp::new(ts).physical(),
- )
+ coprocessor.on_update_safe_ts(core.region_id, ts, ts)
}
}
}
@@ -1261,9 +1253,7 @@ impl RegionReadProgress {
}
}
}
- let self_phy_ts = TimeStamp::new(self.safe_ts()).physical();
- let leader_phy_ts = TimeStamp::new(rs.get_safe_ts()).physical();
- coprocessor.on_update_safe_ts(leader_info.region_id, self_phy_ts, leader_phy_ts)
+ coprocessor.on_update_safe_ts(leader_info.region_id, self.safe_ts(), rs.get_safe_ts())
}
// whether the provided `LeaderInfo` is same as ours
core.leader_info.leader_term == leader_info.term
