From 4a117157b9feff01f7c57109b6393a57a49d714a Mon Sep 17 00:00:00 2001 From: evenyag Date: Tue, 23 Aug 2022 17:39:53 +0800 Subject: [PATCH] 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 | 2 +- src/cmd/src/datanode.rs | 2 +- src/datanode/src/datanode.rs | 2 +- src/storage/src/region/tests/flush.rs | 2 ++ src/storage/src/region/writer.rs | 9 +++++++-- 5 files changed, 12 insertions(+), 5 deletions(-) diff --git a/config/datanode.example.toml b/config/datanode.example.toml index 0a29d6ce40a..30760a108fb 100644 --- a/config/datanode.example.toml +++ b/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 diff --git a/src/cmd/src/datanode.rs b/src/cmd/src/datanode.rs index 40d66e20556..34e93b1e39f 100644 --- a/src/cmd/src/datanode.rs +++ b/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 { diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs index 78f817a5e38..1ae4b0c56fc 100644 --- a/src/datanode/src/datanode.rs +++ b/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(), } } diff --git a/src/storage/src/region/tests/flush.rs b/src/storage/src/region/tests/flush.rs index 50d49957837..195459c8c79 100644 --- a/src/storage/src/region/tests/flush.rs +++ b/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(); diff --git a/src/storage/src/region/writer.rs b/src/storage/src/region/writer.rs index 69ed3d5490f..3a6b3a62221 100644 --- a/src/storage/src/region/writer.rs +++ b/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 {