fix: obkv wal open (#1117)

## Rationale
In some cases(such as `Manifest::do_snapshot`), `actual_next_sequence`
maybe less than `start_sequence`, and it will check this then return
error in OBKV WAl.

In this pr, I refactor this open logic.

## Detailed Changes
Refactor wal open logic to allow `actual_next_sequence` less than
`start_sequence`.

## Test Plan
Test by exist tests.

Author: Rachelint

wal/src/kv_encoder.rs

@@ -263,7 +263,7 @@ pub struct MetaKeyEncoder {
 
 #[derive(Clone, Debug)]
 pub struct MetaKey {
-    pub region_id: u64,
+    pub table_id: u64,
 }
 
 impl MetaKeyEncoder {
@@ -291,7 +291,7 @@ impl Encoder<MetaKey> for MetaKeyEncoder {
         buf.try_put_u8(self.namespace as u8)
             .context(EncodeMetaKey)?;
         buf.try_put_u8(self.key_type as u8).context(EncodeMetaKey)?;
-        buf.try_put_u64(meta_key.region_id).context(EncodeMetaKey)?;
+        buf.try_put_u64(meta_key.table_id).context(EncodeMetaKey)?;
         buf.try_put_u8(self.version).context(EncodeMetaKey)?;
 
         Ok(())
@@ -326,7 +326,7 @@ impl Decoder<MetaKey> for MetaKeyEncoder {
             }
         );
 
-        let region_id = buf.try_get_u64().context(DecodeMetaKey)?;
+        let table_id = buf.try_get_u64().context(DecodeMetaKey)?;
 
         // check version
         let version = buf.try_get_u8().context(DecodeMetaKey)?;
@@ -338,7 +338,7 @@ impl Decoder<MetaKey> for MetaKeyEncoder {
             }
         );
 
-        Ok(MetaKey { region_id })
+        Ok(MetaKey { table_id })
     }
 }
 

wal/src/rocks_impl/manager.rs

@@ -134,7 +134,7 @@ impl TableUnit {
                 .context(Delete)?;
 
             // Update the max sequence number.
-            let meta_key = MetaKey { region_id: self.id };
+            let meta_key = MetaKey { table_id: self.id };
             let meta_value = MaxSeqMetaValue { max_seq };
             let (mut meta_key_buf, mut meta_value_buf) = (BytesMut::new(), BytesMut::new());
             self.max_seq_meta_encoding
@@ -440,7 +440,7 @@ impl RocksImpl {
         &self,
         table_max_seqs: &mut HashMap<TableId, SequenceNumber>,
     ) -> Result<()> {
-        let meta_key = MetaKey { region_id: 0 };
+        let meta_key = MetaKey { table_id: 0 };
         let mut start_boundary_key_buf = BytesMut::new();
         self.max_seq_meta_encoding
             .encode_key(&mut start_boundary_key_buf, &meta_key)?;
@@ -462,9 +462,24 @@ impl RocksImpl {
 
             let meta_key = self.max_seq_meta_encoding.decode_key(iter.key())?;
             let meta_value = self.max_seq_meta_encoding.decode_value(iter.value())?;
+            #[rustfmt::skip]
+            // FIXME: In some cases, the `flushed sequence` 
+            // may be greater than the `actual last sequence of written logs`.
+            // 
+            // Such as following case:
+            //  + Write wal logs failed(last sequence stored in memory will increase when write failed).
+            //  + Get last sequence from memory(greater then actual last sequence now).
+            //  + Mark the got last sequence as flushed sequence.
             table_max_seqs
-                .entry(meta_key.region_id)
+                .entry(meta_key.table_id)
                 .and_modify(|v| {
+                    if meta_value.max_seq > *v {
+                        warn!(
+                            "RocksDB WAL found flushed_seq greater than actual_last_sequence, 
+                        flushed_sequence:{}, actual_last_sequence:{}, table_id:{}",
+                            meta_value.max_seq, *v, meta_key.table_id
+                        );
+                    }
                     *v = meta_value.max_seq.max(*v);
                 })
                 .or_insert(meta_value.max_seq);

wal/src/table_kv_impl/table_unit.rs

@@ -16,7 +16,7 @@ use std::{
 use bytes_ext::BytesMut;
 use common_types::table::TableId;
 use generic_error::{BoxError, GenericError};
-use log::debug;
+use log::{debug, warn};
 use macros::define_result;
 use runtime::{self, Runtime};
 use snafu::{ensure, Backtrace, OptionExt, ResultExt, Snafu};
@@ -462,16 +462,22 @@ impl TableUnit {
                 region_id,
                 table_id,
             )? {
-                ensure!(
-                    sequence + 1 >= start_sequence,
-                    LoadLastSequence {
-                        table_id,
-                        msg: format!(
-                            "found last sequence, but last sequence + 1 < start sequence,
-                            last sequence:{sequence}, start sequence:{start_sequence}",
-                        )
-                    }
-                );
+                #[rustfmt::skip]
+                // FIXME: In some cases, the `flushed sequence`
+                // may be greater than the `actual last sequence of written logs`.
+                //
+                // Such as following case:
+                //  + Write wal logs failed(last sequence stored in memory will increase when write failed). 
+                //  + Get last sequence from memory(greater then actual last sequence now). 
+                //  + Mark the got last sequence as flushed sequence.
+                let actual_next_sequence = sequence + 1;
+                if actual_next_sequence < start_sequence {
+                    warn!("TableKv WAL found start_sequence greater than actual_next_sequence, 
+                    start_sequence:{start_sequence}, actual_next_sequence:{actual_next_sequence}, table_id:{table_id}, region_id:{region_id}");
+
+                    break;
+                }
+
                 return Ok(sequence);
             }
         }