refactor(meta_store): include keys in snapshot list response (risingw…

…avelabs#6632)

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

src/meta/src/storage/etcd_meta_store.rs

@@ -122,7 +122,7 @@ struct ListViewer {
 }
 
 impl SnapshotViewer for ListViewer {
-    type Output = Vec<Vec<u8>>;
+    type Output = Vec<(Vec<u8>, Vec<u8>)>;
 
     type OutputFuture<'a> = impl Future<Output = MetaStoreResult<(i64, Self::Output)>> + 'a;
 
@@ -141,15 +141,19 @@ impl SnapshotViewer for ListViewer {
                     "Etcd response missing header"
                 )));
             };
-            let value = res.kvs().iter().map(|kv| kv.value().to_vec()).collect();
-            Ok((new_revision, value))
+            let kv = res
+                .kvs()
+                .iter()
+                .map(|kv| (kv.key().to_vec(), kv.value().to_vec()))
+                .collect();
+            Ok((new_revision, kv))
         }
     }
 }
 
 #[async_trait]
 impl Snapshot for EtcdSnapshot {
-    async fn list_cf(&self, cf: &str) -> MetaStoreResult<Vec<Vec<u8>>> {
+    async fn list_cf(&self, cf: &str) -> MetaStoreResult<Vec<(Vec<u8>, Vec<u8>)>> {
         let view = ListViewer {
             key: encode_etcd_key(cf, &[]),
         };

src/meta/src/storage/mem_meta_store.rs

@@ -51,9 +51,9 @@ impl MemStoreInner {
 #[async_trait]
 impl Snapshot for MemSnapshot {
     #[inline(always)]
-    async fn list_cf(&self, cf: &str) -> MetaStoreResult<Vec<Value>> {
+    async fn list_cf(&self, cf: &str) -> MetaStoreResult<Vec<(Key, Value)>> {
         Ok(match self.0.cf_ref(cf) {
-            Some(cf) => cf.values().cloned().collect(),
+            Some(cf) => cf.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
             None => vec![],
         })
     }

src/meta/src/storage/meta_store.rs

@@ -22,7 +22,7 @@ pub const DEFAULT_COLUMN_FAMILY: &str = "default";
 
 #[async_trait]
 pub trait Snapshot: Sync + Send + 'static {
-    async fn list_cf(&self, cf: &str) -> MetaStoreResult<Vec<Vec<u8>>>;
+    async fn list_cf(&self, cf: &str) -> MetaStoreResult<Vec<(Vec<u8>, Vec<u8>)>>;
     async fn get_cf(&self, cf: &str, key: &[u8]) -> MetaStoreResult<Vec<u8>>;
 }
 
@@ -38,7 +38,8 @@ pub trait MetaStore: Clone + Sync + Send + 'static {
     async fn txn(&self, trx: Transaction) -> MetaStoreResult<()>;
 
     async fn list_cf(&self, cf: &str) -> MetaStoreResult<Vec<Vec<u8>>> {
-        self.snapshot().await.list_cf(cf).await
+        let kvs = self.snapshot().await.list_cf(cf).await?;
+        Ok(kvs.into_iter().map(|(_k, v)| v).collect())
     }
 
     async fn get_cf(&self, cf: &str, key: &[u8]) -> MetaStoreResult<Vec<u8>> {