refactor(storage): remove enable_test_epoch and modify epochs used by all unit tests to be realistic

src/common/src/util/epoch.rs

@@ -151,8 +151,8 @@ impl EpochPair {
     }
 
     pub fn inc(&mut self) {
-        self.curr += 1;
-        self.prev += 1;
+        self.prev = self.curr;
+        self.curr += 1 << 16;
     }
 
     pub fn inc_for_test(&mut self, inc_by: u64) {
@@ -162,8 +162,131 @@ impl EpochPair {
     }
 
     pub fn new_test_epoch(curr: u64) -> Self {
-        assert!(curr > 0);
-        Self::new(curr, curr - 1)
+        assert!(curr > 65535);
+        Self::new(curr, curr - TestEpoch::new_without_offset(1).as_u64())
+    }
+}
+
+/// The `TestEpoch` struct is used in unit tests to provide consistent logic similar to a normal epoch.
+/// It ensures that the lower 16 bits are always zero and any addition or subtraction operations are only applied to the upper 48 bits.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Default, Debug, PartialOrd, Ord)]
+pub struct TestEpoch {
+    epoch_with_gap: TestEpochWithGap,
+}
+
+impl TestEpoch {
+    pub fn new_without_offset(epoch: u64) -> Self {
+        Self {
+            epoch_with_gap: TestEpochWithGap::new(epoch * (1 << EPOCH_PHYSICAL_SHIFT_BITS), 0),
+        }
+    }
+
+    pub fn new(epoch: u64, spill_offset: u16) -> Self {
+        Self {
+            epoch_with_gap: TestEpochWithGap::new(
+                epoch * (1 << EPOCH_PHYSICAL_SHIFT_BITS),
+                spill_offset,
+            ),
+        }
+    }
+
+    pub fn inc(&mut self) {
+        self.epoch_with_gap.inc();
+    }
+
+    pub fn next_epoch(&self) -> TestEpoch {
+        Self {
+            epoch_with_gap: TestEpochWithGap::new(
+                self.epoch_with_gap.0 + (1 << EPOCH_PHYSICAL_SHIFT_BITS),
+                0,
+            ),
+        }
+    }
+
+    pub fn prev_epoch(&self) -> TestEpoch {
+        Self {
+            epoch_with_gap: TestEpochWithGap::new(
+                self.epoch_with_gap.0 - (1 << EPOCH_PHYSICAL_SHIFT_BITS),
+                0,
+            ),
+        }
+    }
+
+    pub fn sub(&mut self) {
+        self.epoch_with_gap.sub();
+    }
+
+    pub fn inc_by(&mut self, e: u64) {
+        self.epoch_with_gap.inc_by(e);
+    }
+
+    pub fn pure_epoch(&self) -> u64 {
+        self.epoch_with_gap.pure_epoch()
+    }
+
+    pub fn as_u64(&self) -> u64 {
+        self.epoch_with_gap._as_u64()
+    }
+}
+
+/// This structure is the same as `EpochWithGap` in the storage crate,
+/// but it is mocked in the common crate to prevent circular dependencies between the common and storage crates,
+/// as it is needed for all unit tests to use.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Default, Debug, PartialOrd, Ord)]
+pub struct TestEpochWithGap(u64);
+
+impl TestEpochWithGap {
+    #[allow(unused_variables)]
+    pub fn new(epoch: u64, spill_offset: u16) -> Self {
+        // We only use 48 high bit to store epoch and use 16 low bit to store spill offset. But for MAX epoch,
+        // we still keep `u64::MAX` because we have use it in delete range and persist this value to sstable files.
+        //  So for compatibility, we must skip checking it for u64::MAX. See bug description in https://github.com/risingwavelabs/risingwave/issues/13717
+        if is_max_epoch(epoch) {
+            TestEpochWithGap::new_max_epoch()
+        } else {
+            debug_assert!((epoch & EPOCH_SPILL_TIME_MASK) == 0);
+            TestEpochWithGap(epoch + spill_offset as u64)
+        }
+    }
+
+    pub fn new_from_epoch(epoch: u64) -> Self {
+        TestEpochWithGap::new(epoch, 0)
+    }
+
+    pub fn new_min_epoch() -> Self {
+        TestEpochWithGap(0)
+    }
+
+    pub fn new_max_epoch() -> Self {
+        TestEpochWithGap(u64::MAX)
+    }
+
+    // return the epoch_with_gap(epoch + spill_offset)
+    pub(crate) fn _as_u64(&self) -> u64 {
+        self.0
+    }
+
+    // return the pure epoch without spill offset
+    pub fn pure_epoch(&self) -> u64 {
+        self.0 & !EPOCH_SPILL_TIME_MASK
+    }
+
+    pub fn offset(&self) -> u64 {
+        self.0 & EPOCH_SPILL_TIME_MASK
+    }
+
+    pub fn inc(&mut self) {
+        self.0 += 1 << EPOCH_PHYSICAL_SHIFT_BITS;
+    }
+
+    pub fn sub(&mut self) {
+        if self.0 > (1 << EPOCH_PHYSICAL_SHIFT_BITS) {
+            self.0 -= 1 << EPOCH_PHYSICAL_SHIFT_BITS;
+        }
+    }
+
+    pub fn inc_by(&mut self, e: u64) {
+        self.0 += e << EPOCH_PHYSICAL_SHIFT_BITS;
     }
 }
 

src/compute/tests/cdc_tests.rs

@@ -29,6 +29,7 @@ use risingwave_batch::executor::{Executor as BatchExecutor, RowSeqScanExecutor,
 use risingwave_common::array::{Array, ArrayBuilder, DataChunk, Op, StreamChunk, Utf8ArrayBuilder};
 use risingwave_common::catalog::{ColumnDesc, ColumnId, ConflictBehavior, Field, Schema, TableId};
 use risingwave_common::types::{Datum, JsonbVal};
+use risingwave_common::util::epoch::TestEpoch;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_connector::source::cdc::external::mock_external_table::MockExternalTableReader;
 use risingwave_connector::source::cdc::external::{
@@ -290,7 +291,7 @@ async fn test_cdc_backfill() -> StreamResult<()> {
     let stream_chunk2 = create_stream_chunk(chunk2_datums, &chunk_schema);
 
     // The first barrier
-    let curr_epoch = 11;
+    let mut curr_epoch = TestEpoch::new_without_offset(11);
     let mut splits = HashMap::new();
     splits.insert(
         actor_id,
@@ -307,7 +308,7 @@ async fn test_cdc_backfill() -> StreamResult<()> {
         })],
     );
     let init_barrier =
-        Barrier::new_test_barrier(curr_epoch).with_mutation(Mutation::Add(AddMutation {
+        Barrier::new_test_barrier(curr_epoch.as_u64()).with_mutation(Mutation::Add(AddMutation {
             adds: HashMap::new(),
             added_actors: HashSet::new(),
             splits,
@@ -323,7 +324,7 @@ async fn test_cdc_backfill() -> StreamResult<()> {
             epoch,
             mutation: Some(_),
             ..
-        }) if epoch.curr == curr_epoch
+        }) if epoch.curr == curr_epoch.as_u64()
     ));
 
     // start the stream pipeline src -> backfill -> mview
@@ -332,17 +333,19 @@ async fn test_cdc_backfill() -> StreamResult<()> {
     // ingest data and barrier
     let interval = Duration::from_millis(10);
     tx.push_chunk(stream_chunk1);
-
     tokio::time::sleep(interval).await;
-    tx.push_barrier(curr_epoch + 1, false);
+    curr_epoch.inc();
+    tx.push_barrier(curr_epoch.as_u64(), false);
 
     tx.push_chunk(stream_chunk2);
 
     tokio::time::sleep(interval).await;
-    tx.push_barrier(curr_epoch + 2, false);
+    curr_epoch.inc();
+    tx.push_barrier(curr_epoch.as_u64(), false);
 
     tokio::time::sleep(interval).await;
-    tx.push_barrier(curr_epoch + 3, true);
+    curr_epoch.inc();
+    tx.push_barrier(curr_epoch.as_u64(), true);
 
     // scan the final result of the mv table
     let column_descs = vec![

src/compute/tests/integration_tests.rs

@@ -37,7 +37,7 @@ use risingwave_common::row::OwnedRow;
 use risingwave_common::system_param::local_manager::LocalSystemParamsManager;
 use risingwave_common::test_prelude::DataChunkTestExt;
 use risingwave_common::types::{DataType, IntoOrdered};
-use risingwave_common::util::epoch::EpochPair;
+use risingwave_common::util::epoch::{EpochPair, TestEpoch};
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_connector::source::SourceCtrlOpts;
@@ -274,9 +274,9 @@ async fn test_table_materialize() -> StreamResult<()> {
     assert!(result.is_none());
 
     // Send a barrier to start materialized view.
-    let mut curr_epoch = 1919;
+    let mut curr_epoch = TestEpoch::new_without_offset(1919);
     barrier_tx
-        .send(Barrier::new_test_barrier(curr_epoch))
+        .send(Barrier::new_test_barrier(curr_epoch.as_u64()))
         .unwrap();
 
     assert!(matches!(
@@ -285,17 +285,17 @@ async fn test_table_materialize() -> StreamResult<()> {
             epoch,
             mutation: None,
             ..
-        }) if epoch.curr == curr_epoch
+        }) if epoch.curr == curr_epoch.as_u64()
     ));
 
-    curr_epoch += 1;
+    curr_epoch.inc();
     let barrier_tx_clone = barrier_tx.clone();
     tokio::spawn(async move {
         let mut stream = insert.execute();
         let _ = stream.next().await.unwrap()?;
         // Send a barrier and poll again, should write changes to storage.
         barrier_tx_clone
-            .send(Barrier::new_test_barrier(curr_epoch))
+            .send(Barrier::new_test_barrier(curr_epoch.as_u64()))
             .unwrap();
         Ok::<_, RwError>(())
     });
@@ -325,7 +325,7 @@ async fn test_table_materialize() -> StreamResult<()> {
             epoch,
             mutation: None,
             ..
-        }) if epoch.curr == curr_epoch
+        }) if epoch.curr == curr_epoch.as_u64()
     ));
 
     // Scan the table again, we are able to get the data now!
@@ -370,14 +370,14 @@ async fn test_table_materialize() -> StreamResult<()> {
         0,
     ));
 
-    curr_epoch += 1;
+    curr_epoch.inc();
     let barrier_tx_clone = barrier_tx.clone();
     tokio::spawn(async move {
         let mut stream = delete.execute();
         let _ = stream.next().await.unwrap()?;
         // Send a barrier and poll again, should write changes to storage.
         barrier_tx_clone
-            .send(Barrier::new_test_barrier(curr_epoch))
+            .send(Barrier::new_test_barrier(curr_epoch.as_u64()))
             .unwrap();
         Ok::<_, RwError>(())
     });
@@ -404,7 +404,7 @@ async fn test_table_materialize() -> StreamResult<()> {
             epoch,
             mutation: None,
             ..
-        }) if epoch.curr == curr_epoch
+        }) if epoch.curr == curr_epoch.as_u64()
     ));
 
     // Scan the table again, we are able to see the deletion now!
@@ -463,7 +463,7 @@ async fn test_row_seq_scan() -> Result<()> {
         vec![0, 1, 2],
     );
 
-    let mut epoch = EpochPair::new_test_epoch(1);
+    let mut epoch = EpochPair::new_test_epoch(TestEpoch::new_without_offset(1).as_u64());
     state.init_epoch(epoch);
     state.insert(OwnedRow::new(vec![
         Some(1_i32.into()),

src/connector/src/sink/log_store.rs

@@ -531,6 +531,7 @@ mod tests {
     use std::task::Poll;
 
     use futures::{FutureExt, TryFuture};
+    use risingwave_common::util::epoch::TestEpoch;
     use tokio::sync::oneshot;
     use tokio::sync::oneshot::Receiver;
 
@@ -639,33 +640,33 @@ mod tests {
     #[tokio::test]
     async fn test_future_delivery_manager_compress_chunk() {
         let mut manager = DeliveryFutureManager::new(10);
-        let epoch1 = 233;
+        let epoch1 = TestEpoch::new_without_offset(233);
         let chunk_id1 = 1;
         let chunk_id2 = chunk_id1 + 1;
         let chunk_id3 = chunk_id2 + 1;
         let (tx1_1, rx1_1) = oneshot::channel();
         let (tx1_2, rx1_2) = oneshot::channel();
         let (tx1_3, rx1_3) = oneshot::channel();
-        let epoch2 = epoch1 + 1;
+        let epoch2 = TestEpoch::new_without_offset(234);
         let (tx2_1, rx2_1) = oneshot::channel();
         assert!(!manager
-            .start_write_chunk(epoch1, chunk_id1)
+            .start_write_chunk(epoch1.as_u64(), chunk_id1)
             .add_future_may_await(to_test_future(rx1_1))
             .await
             .unwrap());
         assert!(!manager
-            .start_write_chunk(epoch1, chunk_id2)
+            .start_write_chunk(epoch1.as_u64(), chunk_id2)
             .add_future_may_await(to_test_future(rx1_2))
             .await
             .unwrap());
         assert!(!manager
-            .start_write_chunk(epoch1, chunk_id3)
+            .start_write_chunk(epoch1.as_u64(), chunk_id3)
             .add_future_may_await(to_test_future(rx1_3))
             .await
             .unwrap());
-        manager.add_barrier(epoch1);
+        manager.add_barrier(epoch1.as_u64());
         assert!(!manager
-            .start_write_chunk(epoch2, chunk_id1)
+            .start_write_chunk(epoch2.as_u64(), chunk_id1)
             .add_future_may_await(to_test_future(rx2_1))
             .await
             .unwrap());
@@ -688,7 +689,7 @@ mod tests {
             assert_eq!(
                 next_truncate_offset.await.unwrap(),
                 TruncateOffset::Chunk {
-                    epoch: epoch1,
+                    epoch: epoch1.as_u64(),
                     chunk_id: chunk_id2
                 }
             );
@@ -706,14 +707,16 @@ mod tests {
             // Emit barrier though later chunk has finished.
             assert_eq!(
                 next_truncate_offset.await.unwrap(),
-                TruncateOffset::Barrier { epoch: epoch1 }
+                TruncateOffset::Barrier {
+                    epoch: epoch1.as_u64()
+                }
             );
         }
         assert_eq!(manager.future_count, 1);
         assert_eq!(
             manager.next_truncate_offset().await.unwrap(),
             TruncateOffset::Chunk {
-                epoch: epoch2,
+                epoch: epoch2.as_u64(),
                 chunk_id: chunk_id1
             }
         );