feat: Implement delete operation for WAL based on local storage (#1566)

## Rationale

Currently the WAL based on the local disk does not support the delete
function. This PR implements that functionality.

This is a follow-up task of #1552 and #1556.

## Detailed Changes

1. For each `Segment`, add a hashmap to record the minimum and maximum
sequence numbers of all tables within that segment. During `delete` and
`write` operations, this hashmap will be updated. During read
operations, logs will be filtered based on this hashmap.

2. During the `delete` operation, based on the aforementioned hashmap,
if all logs of all tables in a read-only segment (a segment that is not
currently being written to) are marked as deleted, the segment file will
be physically deleted from the disk.

## Test Plan

Unit test, TSBS and running a script locally that repeatedly inserts
data, forcibly kills, and restarts the database process to test
persistence.

src/analytic_engine/src/instance/write.rs

@@ -531,21 +531,6 @@ impl<'a> Writer<'a> {
                 e
             })?;
 
-        // When wal is disabled, there is no need to do this check.
-        if !self.instance.disable_wal {
-            // NOTE: Currently write wal will only increment seq by one,
-            // this may change in future.
-            let last_seq = table_data.last_sequence();
-            if sequence != last_seq + 1 {
-                warn!(
-                    "Sequence must be consecutive, table:{}, table_id:{}, last_sequence:{}, wal_sequence:{}",
-                    table_data.name,table_data.id,
-                    table_data.last_sequence(),
-                    sequence
-                );
-            }
-        }
-
         debug!(
             "Instance write finished, update sequence, table:{}, table_id:{} last_sequence:{}",
             table_data.name, table_data.id, sequence

src/wal/Cargo.toml

@@ -47,6 +47,7 @@ name = "read_write"
 required-features = ["wal-message-queue", "wal-table-kv", "wal-rocksdb", "wal-local-storage"]
 
 [dependencies]
+anyhow = { workspace = true }
 async-trait = { workspace = true }
 bytes_ext = { workspace = true }
 chrono = { workspace = true }

src/wal/src/local_storage_impl/config.rs

@@ -18,17 +18,18 @@
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
 pub struct LocalStorageConfig {
-    pub path: String,
-    pub max_segment_size: usize,
+    pub data_dir: String,
+    pub segment_size: usize,
     pub cache_size: usize,
 }
 
 impl Default for LocalStorageConfig {
     fn default() -> Self {
         Self {
-            path: "/tmp/horaedb".to_string(),
-            max_segment_size: 64 * 1024 * 1024, // 64MB
+            data_dir: "/tmp/horaedb".to_string(),
+            segment_size: 64 * 1024 * 1024, // 64MB
             cache_size: 3,
         }
     }