fix: Fix replay sequence and wal dir (#196)

* fix: Fix replay include flushed data

Replay should starts from flushed_sequence + 1

* fix: Move default wal path to `/tmp/greptimedb`

config/datanode.example.toml

@@ -1,6 +1,6 @@
 http_addr = '0.0.0.0:3000'
 rpc_addr = '0.0.0.0:3001'
-wal_dir = '/tmp/wal'
+wal_dir = '/tmp/greptimedb/wal'
 
 mysql_addr = '0.0.0.0:3306'
 mysql_runtime_size = 4

src/cmd/src/datanode.rs

@@ -103,7 +103,7 @@ mod tests {
         let options: DatanodeOptions = cmd.try_into().unwrap();
         assert_eq!("0.0.0.0:3000".to_string(), options.http_addr);
         assert_eq!("0.0.0.0:3001".to_string(), options.rpc_addr);
-        assert_eq!("/tmp/wal".to_string(), options.wal_dir);
+        assert_eq!("/tmp/greptimedb/wal".to_string(), options.wal_dir);
         assert_eq!("0.0.0.0:3306".to_string(), options.mysql_addr);
         assert_eq!(4, options.mysql_runtime_size);
         match options.storage {

src/datanode/src/datanode.rs

@@ -37,7 +37,7 @@ impl Default for DatanodeOptions {
             rpc_addr: "0.0.0.0:3001".to_string(),
             mysql_addr: "0.0.0.0:3306".to_string(),
             mysql_runtime_size: 2,
-            wal_dir: "/tmp/wal".to_string(),
+            wal_dir: "/tmp/greptimedb/wal".to_string(),
             storage: ObjectStoreConfig::default(),
         }
     }

src/storage/src/region/tests/flush.rs

@@ -78,6 +78,8 @@ impl FlushStrategy for FlushSwitch {
     }
 }
 
+// TODO(yingwen): Test replay after flush.
+
 #[tokio::test]
 async fn test_flush_and_stall() {
     common_telemetry::init_default_ut_logging();

src/storage/src/region/writer.rs

@@ -210,7 +210,9 @@ impl WriterInner {
             // Data after flushed sequence need to be recovered.
             flushed_sequence = version_control.current().flushed_sequence();
             last_sequence = flushed_sequence;
-            let mut stream = writer_ctx.wal.read_from_wal(flushed_sequence).await?;
+            // Read starts from the first entry after last flushed entry, so the start sequence
+            // should be flushed_sequence + 1.
+            let mut stream = writer_ctx.wal.read_from_wal(flushed_sequence + 1).await?;
             while let Some((req_sequence, _header, request)) = stream.try_next().await? {
                 if let Some(request) = request {
                     num_requests += 1;
@@ -222,11 +224,14 @@ impl WriterInner {
                         last_sequence = req_sequence;
                     } else {
                         logging::error!(
-                            "Sequence should not decrease during replay, found {} < {}, region_id: {}, region_name: {}",
+                            "Sequence should not decrease during replay, found {} <= {}, \
+                            region_id: {}, region_name: {}, flushed_sequence: {}, num_requests: {}",
                             req_sequence,
                             last_sequence,
                             writer_ctx.shared.id,
                             writer_ctx.shared.name,
+                            flushed_sequence,
+                            num_requests,
                         );
 
                         error::SequenceNotMonotonicSnafu {