fix: correct the range behavior in Memory KvBackend

src/common/meta/src/kv_backend/memory.rs

@@ -96,6 +96,14 @@ impl<T: ErrorExt + Send + Sync + 'static> KvBackend for MemoryKvBackend<T> {
 
         let iter: Box<dyn Iterator<Item = (&Vec<u8>, &Vec<u8>)>> = if range_end.is_empty() {
             Box::new(kvs.get_key_value(&key).into_iter())
+        } else if range_end.len() == 1 && range_end[0] == 0 {
+            // If both key and range_end are ‘\0’, then range represents all keys.
+            if key.len() == 1 && key[0] == 0 {
+                Box::new(kvs.iter())
+            } else {
+                // If range_end is ‘\0’, the range is all keys greater than or equal to the key argument.
+                Box::new(kvs.range(key..))
+            }
         } else {
             Box::new(kvs.range(key..range_end))
         };
@@ -504,6 +512,43 @@ mod tests {
         assert_eq!(b"val1", resp.kvs[0].value());
     }
 
+    #[tokio::test]
+    async fn test_range_2() {
+        let kv = MemoryKvBackend::<Error>::new();
+
+        kv.put(PutRequest::new().with_key("atest").with_value("value"))
+            .await
+            .unwrap();
+
+        kv.put(PutRequest::new().with_key("test").with_value("value"))
+            .await
+            .unwrap();
+
+        // If both key and range_end are ‘\0’, then range represents all keys.
+        let result = kv
+            .range(RangeRequest::new().with_range(b"\0".to_vec(), b"\0".to_vec()))
+            .await
+            .unwrap();
+
+        assert_eq!(result.kvs.len(), 2);
+
+        // If range_end is ‘\0’, the range is all keys greater than or equal to the key argument.
+        let result = kv
+            .range(RangeRequest::new().with_range(b"a".to_vec(), b"\0".to_vec()))
+            .await
+            .unwrap();
+
+        assert_eq!(result.kvs.len(), 2);
+
+        let result = kv
+            .range(RangeRequest::new().with_range(b"b".to_vec(), b"\0".to_vec()))
+            .await
+            .unwrap();
+
+        assert_eq!(result.kvs.len(), 1);
+        assert_eq!(result.kvs[0].key, b"test");
+    }
+
     #[tokio::test]
     async fn test_batch_get() {
         let kv_store = mock_mem_store_with_data().await;