fix(compaction): fix max epoch bug of tombstone

src/common/src/util/epoch.rs

@@ -118,6 +118,13 @@ pub const EPOCH_AVAILABLE_BITS: u64 = 16;
 pub const MAX_SPILL_TIMES: u16 = ((1 << EPOCH_AVAILABLE_BITS) - 1) as u16;
 pub const EPOCH_MASK: u64 = (1 << EPOCH_AVAILABLE_BITS) - 1;
 pub const MAX_EPOCH: u64 = u64::MAX & !EPOCH_MASK;
+
+pub fn is_max_epoch(epoch: u64) -> bool {
+    epoch >= MAX_EPOCH
+}
+pub fn is_compatibility_epoch(epoch: u64) -> bool {
+    epoch == MAX_EPOCH
+}
 impl From<u64> for Epoch {
     fn from(epoch: u64) -> Self {
         Self(epoch)

src/ctl/src/cmd_impl/hummock/list_kv.rs

@@ -15,7 +15,7 @@
 use core::ops::Bound::Unbounded;
 
 use risingwave_common::catalog::TableId;
-use risingwave_common::util::epoch::MAX_EPOCH;
+use risingwave_common::util::epoch::is_max_epoch;
 use risingwave_storage::hummock::CachePolicy;
 use risingwave_storage::store::{PrefetchOptions, ReadOptions, StateStoreReadExt};
 
@@ -31,8 +31,8 @@ pub async fn list_kv(
     let hummock = context
         .hummock_store(HummockServiceOpts::from_env(data_dir)?)
         .await?;
-    if epoch == MAX_EPOCH {
-        tracing::info!("using MAX_EPOCH as epoch");
+    if is_max_epoch(epoch) {
+        tracing::info!("using MAX EPOCH as epoch");
     }
     let scan_result = {
         let range = (Unbounded, Unbounded);

src/ctl/src/cmd_impl/hummock/list_version_deltas.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_common::util::epoch::MAX_EPOCH;
+use risingwave_hummock_sdk::HummockEpoch;
 
 use crate::CtlContext;
 
@@ -23,7 +23,7 @@ pub async fn list_version_deltas(
 ) -> anyhow::Result<()> {
     let meta_client = context.meta_client().await?;
     let resp = meta_client
-        .list_version_deltas(start_id, num_epochs, MAX_EPOCH)
+        .list_version_deltas(start_id, num_epochs, HummockEpoch::MAX)
         .await?;
     println!("{:#?}", resp.version_deltas);
     Ok(())

src/ctl/src/cmd_impl/hummock/pause_resume.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_common::util::epoch::MAX_EPOCH;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionUpdateExt;
+use risingwave_hummock_sdk::HummockEpoch;
 
 use crate::CtlContext;
 
@@ -62,7 +62,7 @@ pub async fn replay_version(context: &CtlContext) -> anyhow::Result<()> {
     let mut current_delta_id = base_version.id + 1;
     loop {
         let deltas = meta_client
-            .list_version_deltas(current_delta_id, delta_fetch_size, MAX_EPOCH)
+            .list_version_deltas(current_delta_id, delta_fetch_size, HummockEpoch::MAX)
             .await
             .unwrap();
         if deltas.version_deltas.is_empty() {

src/ctl/src/lib.rs

@@ -19,7 +19,7 @@ use anyhow::Result;
 use clap::{Args, Parser, Subcommand};
 use cmd_impl::bench::BenchCommands;
 use cmd_impl::hummock::SstDumpArgs;
-use risingwave_common::util::epoch::MAX_EPOCH;
+use risingwave_hummock_sdk::HummockEpoch;
 use risingwave_meta::backup_restore::RestoreOpts;
 use risingwave_pb::meta::update_worker_node_schedulability_request::Schedulability;
 
@@ -160,7 +160,7 @@ enum HummockCommands {
     DisableCommitEpoch,
     /// list all Hummock key-value pairs
     ListKv {
-        #[clap(short, long = "epoch", default_value_t = MAX_EPOCH)]
+        #[clap(short, long = "epoch", default_value_t = HummockEpoch::MAX)]
         epoch: u64,
 
         #[clap(short, long = "table-id")]

src/frontend/src/meta_client.rs

@@ -15,7 +15,6 @@
 use std::collections::HashMap;
 
 use risingwave_common::system_param::reader::SystemParamsReader;
-use risingwave_common::util::epoch::MAX_EPOCH;
 use risingwave_pb::backup_service::MetaSnapshotMetadata;
 use risingwave_pb::catalog::Table;
 use risingwave_pb::common::WorkerNode;
@@ -230,7 +229,7 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
     async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>> {
         // FIXME #8612: there can be lots of version deltas, so better to fetch them by pages and refactor `SysRowSeqScanExecutor` to yield multiple chunks.
         self.0
-            .list_version_deltas(0, u32::MAX, MAX_EPOCH)
+            .list_version_deltas(0, u32::MAX, u64::MAX)
             .await
             .map(|v| v.version_deltas)
     }

src/meta/src/hummock/compaction/mod.rs

@@ -17,7 +17,6 @@
 pub mod compaction_config;
 mod overlap_strategy;
 use risingwave_common::catalog::TableOption;
-use risingwave_common::util::epoch::MAX_EPOCH;
 use risingwave_hummock_sdk::prost_key_range::KeyRangeExt;
 use risingwave_pb::hummock::compact_task::{self, TaskStatus, TaskType};
 
@@ -29,7 +28,9 @@ use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
 use picker::{LevelCompactionPicker, TierCompactionPicker};
-use risingwave_hummock_sdk::{can_concat, CompactionGroupId, HummockCompactionTaskId};
+use risingwave_hummock_sdk::{
+    can_concat, CompactionGroupId, HummockCompactionTaskId, HummockEpoch,
+};
 use risingwave_pb::hummock::compaction_config::CompactionMode;
 use risingwave_pb::hummock::hummock_version::Levels;
 use risingwave_pb::hummock::{CompactTask, CompactionConfig, KeyRange, LevelType};
@@ -130,7 +131,7 @@ impl CompactStatus {
         let compact_task = CompactTask {
             input_ssts: ret.input.input_levels,
             splits: vec![KeyRange::inf()],
-            watermark: MAX_EPOCH,
+            watermark: HummockEpoch::MAX,
             sorted_output_ssts: vec![],
             task_id,
             target_level: target_level_id as u32,

src/meta/src/hummock/metrics_utils.rs

@@ -19,12 +19,11 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use itertools::{enumerate, Itertools};
 use prost::Message;
-use risingwave_common::util::epoch::MAX_EPOCH;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext::{
     object_size_map, BranchedSstInfo, HummockVersionExt,
 };
 use risingwave_hummock_sdk::{
-    CompactionGroupId, HummockContextId, HummockSstableObjectId, HummockVersionId,
+    CompactionGroupId, HummockContextId, HummockEpoch, HummockSstableObjectId, HummockVersionId,
 };
 use risingwave_pb::hummock::hummock_version::Levels;
 use risingwave_pb::hummock::write_limits::WriteLimit;
@@ -345,7 +344,7 @@ pub fn trigger_pin_unpin_snapshot_state(
     {
         metrics.min_pinned_epoch.set(m as i64);
     } else {
-        metrics.min_pinned_epoch.set(MAX_EPOCH as _);
+        metrics.min_pinned_epoch.set(HummockEpoch::MAX as _);
     }
 }
 

src/storage/hummock_sdk/src/lib.rs

@@ -26,7 +26,7 @@ mod key_cmp;
 use std::cmp::Ordering;
 
 pub use key_cmp::*;
-use risingwave_common::util::epoch::{EPOCH_MASK, MAX_EPOCH};
+use risingwave_common::util::epoch::EPOCH_MASK;
 use risingwave_pb::common::{batch_query_epoch, BatchQueryEpoch};
 use risingwave_pb::hummock::SstableInfo;
 
@@ -281,7 +281,7 @@ impl EpochWithGap {
     pub fn new(epoch: u64, spill_offset: u16) -> Self {
         #[cfg(not(feature = "enable_test_epoch"))]
         {
-            debug_assert_eq!(epoch & EPOCH_MASK, 0);
+            debug_assert!(((epoch & EPOCH_MASK) == 0) || (epoch == HummockEpoch::MAX));
             let epoch_with_gap = epoch + spill_offset as u64;
             EpochWithGap(epoch_with_gap)
         }
@@ -300,7 +300,7 @@ impl EpochWithGap {
     }
 
     pub fn new_max_epoch() -> Self {
-        EpochWithGap(MAX_EPOCH)
+        EpochWithGap(HummockEpoch::MAX)
     }
 
     // return the epoch_with_gap(epoch + spill_offset)

src/storage/hummock_test/benches/bench_hummock_iter.rs

@@ -19,8 +19,8 @@ use bytes::Bytes;
 use criterion::{criterion_group, criterion_main, Criterion};
 use futures::{pin_mut, TryStreamExt};
 use risingwave_common::cache::CachePriority;
-use risingwave_common::util::epoch::MAX_EPOCH;
 use risingwave_hummock_sdk::key::TableKey;
+use risingwave_hummock_sdk::HummockEpoch;
 use risingwave_hummock_test::get_notification_client_for_test;
 use risingwave_hummock_test::local_state_store_test_utils::LocalStateStoreTestExt;
 use risingwave_hummock_test::test_utils::TestIngestBatch;
@@ -99,7 +99,7 @@ fn criterion_benchmark(c: &mut Criterion) {
             ))
             .unwrap();
     }
-    hummock_storage.seal_current_epoch(MAX_EPOCH, SealCurrentEpochOptions::for_test());
+    hummock_storage.seal_current_epoch(HummockEpoch::MAX, SealCurrentEpochOptions::for_test());
 
     c.bench_function("bench-hummock-iter", move |b| {
         b.iter(|| {

src/storage/src/hummock/compactor/compactor_runner.rs

@@ -19,14 +19,14 @@ use await_tree::InstrumentAwait;
 use bytes::Bytes;
 use futures::{stream, FutureExt, StreamExt};
 use itertools::Itertools;
-use risingwave_common::util::epoch::MAX_EPOCH;
+use risingwave_common::util::epoch::is_max_epoch;
 use risingwave_hummock_sdk::compact::{
     compact_task_to_string, estimate_memory_for_compact_task, statistics_compact_task,
 };
 use risingwave_hummock_sdk::key::{FullKey, PointRange};
 use risingwave_hummock_sdk::key_range::{KeyRange, KeyRangeCommon};
 use risingwave_hummock_sdk::table_stats::{add_table_stats_map, TableStats, TableStatsMap};
-use risingwave_hummock_sdk::{can_concat, EpochWithGap};
+use risingwave_hummock_sdk::{can_concat, EpochWithGap, HummockEpoch};
 use risingwave_pb::hummock::compact_task::{TaskStatus, TaskType};
 use risingwave_pb::hummock::{BloomFilterType, CompactTask, LevelType};
 use tokio::sync::oneshot::Receiver;
@@ -157,7 +157,7 @@ impl CompactorRunner {
             .context
             .storage_opts
             .compact_iter_recreate_timeout_ms;
-        let mut del_iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
+        let mut del_iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);
 
         for level in &self.compact_task.input_ssts {
             if level.table_infos.is_empty() {
@@ -657,7 +657,7 @@ where
         del_iter.seek(full_key.user_key).await?;
         if !task_config.gc_delete_keys
             && del_iter.is_valid()
-            && del_iter.earliest_epoch() != MAX_EPOCH
+            && !is_max_epoch(del_iter.earliest_epoch())
         {
             sst_builder
                 .add_monotonic_delete(MonotonicDeleteEvent {
@@ -680,7 +680,7 @@ where
 
     let mut last_key = FullKey::default();
     let mut watermark_can_see_last_key = false;
-    let mut user_key_last_delete_epoch = MAX_EPOCH;
+    let mut user_key_last_delete_epoch = HummockEpoch::MAX;
     let mut local_stats = StoreLocalStatistic::default();
 
     // Keep table stats changes due to dropping KV.
@@ -716,7 +716,7 @@ where
             }
             last_key.set(iter_key);
             watermark_can_see_last_key = false;
-            user_key_last_delete_epoch = MAX_EPOCH;
+            user_key_last_delete_epoch = HummockEpoch::MAX;
             if value.is_delete() {
                 local_stats.skip_delete_key_count += 1;
             }
@@ -843,7 +843,7 @@ where
                 sst_builder
                     .add_monotonic_delete(MonotonicDeleteEvent {
                         event_key: extended_largest_user_key,
-                        new_epoch: MAX_EPOCH,
+                        new_epoch: HummockEpoch::MAX,
                     })
                     .await?;
                 break;
@@ -964,7 +964,7 @@ mod tests {
             .cloned()
             .collect_vec();
 
-        let mut iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
+        let mut iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);
         iter.add_concat_iter(sstable_infos, sstable_store);
 
         let ret = CompactionDeleteRangeIterator::new(iter)

src/storage/src/hummock/compactor/shared_buffer_compact.rs

@@ -23,11 +23,10 @@ use itertools::Itertools;
 use risingwave_common::cache::CachePriority;
 use risingwave_common::catalog::TableId;
 use risingwave_common::hash::VirtualNode;
-use risingwave_common::util::epoch::MAX_EPOCH;
 use risingwave_hummock_sdk::compaction_group::StaticCompactionGroupId;
 use risingwave_hummock_sdk::key::{FullKey, PointRange, TableKey, UserKey};
 use risingwave_hummock_sdk::key_range::KeyRange;
-use risingwave_hummock_sdk::{CompactionGroupId, EpochWithGap, LocalSstableInfo};
+use risingwave_hummock_sdk::{CompactionGroupId, EpochWithGap, HummockEpoch, LocalSstableInfo};
 use risingwave_pb::hummock::compact_task;
 use tracing::error;
 
@@ -241,7 +240,7 @@ async fn compact_shared_buffer(
             Box::new(sstable_object_id_manager.clone()),
         );
         let mut forward_iters = Vec::with_capacity(payload.len());
-        let mut del_iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
+        let mut del_iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);
         for imm in &payload {
             forward_iters.push(imm.clone().into_forward_iter());
             del_iter.add_batch_iter(imm.delete_range_iter());
@@ -322,7 +321,7 @@ pub async fn merge_imms_in_memory(
     let mut largest_table_key = Bound::Included(Bytes::new());
 
     let mut imm_iters = Vec::with_capacity(imms.len());
-    let mut del_iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
+    let mut del_iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);
     for imm in imms {
         assert!(
             imm.kv_count() > 0 || imm.has_range_tombstone(),
@@ -394,14 +393,14 @@ pub async fn merge_imms_in_memory(
 
     let mut versions: Vec<(EpochWithGap, HummockValue<Bytes>)> = Vec::new();
 
-    let mut pivot_last_delete_epoch = MAX_EPOCH;
+    let mut pivot_last_delete_epoch = HummockEpoch::MAX;
 
     for ((key, value), epoch) in items {
         assert!(key >= pivot, "key should be in ascending order");
         if key != pivot {
             merged_payload.push((pivot, versions));
             pivot = key;
-            pivot_last_delete_epoch = MAX_EPOCH;
+            pivot_last_delete_epoch = HummockEpoch::MAX;
             versions = vec![];
             let target_extended_user_key =
                 PointRange::from_user_key(UserKey::new(table_id, TableKey(pivot.as_ref())), false);

src/storage/src/hummock/event_handler/hummock_event_handler.rs

@@ -21,7 +21,6 @@ use arc_swap::ArcSwap;
 use await_tree::InstrumentAwait;
 use parking_lot::RwLock;
 use prometheus::core::{AtomicU64, GenericGauge};
-use risingwave_common::util::epoch::MAX_EPOCH;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionUpdateExt;
 use risingwave_hummock_sdk::{info_in_release, HummockEpoch, LocalSstableInfo};
 use risingwave_pb::hummock::version_update_payload::Payload;
@@ -396,7 +395,7 @@ impl HummockEventHandler {
         }
 
         self.sstable_object_id_manager
-            .remove_watermark_object_id(TrackerId::Epoch(MAX_EPOCH));
+            .remove_watermark_object_id(TrackerId::Epoch(HummockEpoch::MAX));
 
         // Notify completion of the Clear event.
         let _ = notifier.send(()).inspect_err(|e| {

src/storage/src/hummock/iterator/backward_user.rs

@@ -277,13 +277,7 @@ impl<I: HummockIterator<Direction = Backward>> BackwardUserIterator<I> {
 impl<I: HummockIterator<Direction = Backward>> BackwardUserIterator<I> {
     /// Creates [`BackwardUserIterator`] with maximum epoch.
     pub(crate) fn for_test(iterator: I, key_range: UserKeyRange) -> Self {
-        Self::with_epoch(
-            iterator,
-            key_range,
-            risingwave_common::util::epoch::MAX_EPOCH,
-            0,
-            None,
-        )
+        Self::with_epoch(iterator, key_range, HummockEpoch::MAX, 0, None)
     }
 
     /// Creates [`BackwardUserIterator`] with maximum epoch.
@@ -292,13 +286,7 @@ impl<I: HummockIterator<Direction = Backward>> BackwardUserIterator<I> {
         key_range: UserKeyRange,
         min_epoch: HummockEpoch,
     ) -> Self {
-        Self::with_epoch(
-            iterator,
-            key_range,
-            risingwave_common::util::epoch::MAX_EPOCH,
-            min_epoch,
-            None,
-        )
+        Self::with_epoch(iterator, key_range, HummockEpoch::MAX, min_epoch, None)
     }
 }
 

src/storage/src/hummock/iterator/concat_delete_range_iterator.rs

@@ -185,7 +185,7 @@ mod tests {
 
     use bytes::Bytes;
     use risingwave_common::catalog::TableId;
-    use risingwave_common::util::epoch::MAX_EPOCH;
+    use risingwave_common::util::epoch::is_max_epoch;
 
     use super::*;
     use crate::hummock::iterator::test_utils::mock_sstable_store;
@@ -252,7 +252,7 @@ mod tests {
             sstable_store,
         );
         concat_iterator.rewind().await.unwrap();
-        assert_eq!(concat_iterator.current_epoch(), MAX_EPOCH);
+        assert!(is_max_epoch(concat_iterator.current_epoch()));
         assert_eq!(
             concat_iterator.next_extended_user_key().left_user_key,
             test_user_key(b"aaaa").as_ref()