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,8 @@ 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 - 65536)
     }
 }
 

src/compute/tests/cdc_tests.rs

@@ -36,7 +36,7 @@ use risingwave_connector::source::cdc::external::{
 };
 use risingwave_connector::source::cdc::{CdcSplitBase, DebeziumCdcSplit, MySqlCdcSplit};
 use risingwave_connector::source::SplitImpl;
-use risingwave_hummock_sdk::to_committed_batch_query_epoch;
+use risingwave_hummock_sdk::{to_committed_batch_query_epoch, EpochWithGap};
 use risingwave_storage::memory::MemoryStateStore;
 use risingwave_storage::table::batch_table::storage_table::StorageTable;
 use risingwave_stream::common::table::state_table::StateTable;
@@ -290,7 +290,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 = EpochWithGap::new_without_offset(11);
     let mut splits = HashMap::new();
     splits.insert(
         actor_id,
@@ -306,13 +306,14 @@ async fn test_cdc_backfill() -> StreamResult<()> {
             _phantom: PhantomData,
         })],
     );
-    let init_barrier =
-        Barrier::new_test_barrier(curr_epoch).with_mutation(Mutation::Add(AddMutation {
+    let init_barrier = Barrier::new_test_barrier(curr_epoch.as_u64_for_test()).with_mutation(
+        Mutation::Add(AddMutation {
             adds: HashMap::new(),
             added_actors: HashSet::new(),
             splits,
             pause: false,
-        }));
+        }),
+    );
 
     tx.send_barrier(init_barrier);
 
@@ -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_for_test()
     ));
 
     // 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_for_test(), 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_for_test(), false);
 
     tokio::time::sleep(interval).await;
-    tx.push_barrier(curr_epoch + 3, true);
+    curr_epoch.inc();
+    tx.push_barrier(curr_epoch.as_u64_for_test(), true);
 
     // scan the final result of the mv table
     let column_descs = vec![

src/compute/tests/integration_tests.rs

@@ -42,7 +42,7 @@ use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_connector::source::SourceCtrlOpts;
 use risingwave_connector::ConnectorParams;
-use risingwave_hummock_sdk::to_committed_batch_query_epoch;
+use risingwave_hummock_sdk::{to_committed_batch_query_epoch, EpochWithGap};
 use risingwave_pb::catalog::StreamSourceInfo;
 use risingwave_pb::plan_common::PbRowFormatType;
 use risingwave_source::connector_test_utils::create_source_desc_builder;
@@ -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 = EpochWithGap::new_without_offset(1919);
     barrier_tx
-        .send(Barrier::new_test_barrier(curr_epoch))
+        .send(Barrier::new_test_barrier(curr_epoch.as_u64_for_test()))
         .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_for_test()
     ));
 
-    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_for_test()))
             .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_for_test()
     ));
 
     // 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_for_test()))
             .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_for_test()
     ));
 
     // Scan the table again, we are able to see the deletion now!
@@ -463,7 +463,8 @@ 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(EpochWithGap::new_without_offset(1).as_u64_for_test());
     state.init_epoch(epoch);
     state.insert(OwnedRow::new(vec![
         Some(1_i32.into()),

src/connector/Cargo.toml

@@ -107,6 +107,7 @@ regex = "1.4"
 reqwest = { version = "0.11", features = ["json"] }
 risingwave_common = { workspace = true }
 risingwave_jni_core = { workspace = true }
+risingwave_hummock_sdk = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
 rust_decimal = "1"

src/connector/src/sink/log_store.rs

@@ -531,6 +531,7 @@ mod tests {
     use std::task::Poll;
 
     use futures::{FutureExt, TryFuture};
+    use risingwave_hummock_sdk::EpochWithGap;
     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 = EpochWithGap::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 = EpochWithGap::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_for_test(), 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_for_test(), 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_for_test(), chunk_id3)
             .add_future_may_await(to_test_future(rx1_3))
             .await
             .unwrap());
-        manager.add_barrier(epoch1);
+        manager.add_barrier(epoch1.as_u64_for_test());
         assert!(!manager
-            .start_write_chunk(epoch2, chunk_id1)
+            .start_write_chunk(epoch2.as_u64_for_test(), 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_for_test(),
                     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_for_test()
+                }
             );
         }
         assert_eq!(manager.future_count, 1);
         assert_eq!(
             manager.next_truncate_offset().await.unwrap(),
             TruncateOffset::Chunk {
-                epoch: epoch2,
+                epoch: epoch2.as_u64_for_test(),
                 chunk_id: chunk_id1
             }
         );