services/redis: Implement Write::append with native support (#1651)

Signed-off-by: Wenyu Huang <huangwenyuu@outlook.com>

src/raw/adapters/kv/api.rs

@@ -104,6 +104,31 @@ pub trait Adapter: Send + Sync + Debug + Unpin + 'static {
         )
         .with_operation("kv::Adapter::blocking_scan"))
     }
+
+    /// Append a key into service
+    async fn append(&self, path: &str, value: &[u8]) -> Result<()> {
+        let _ = path;
+        let _ = value;
+
+        Err(Error::new(
+            ErrorKind::Unsupported,
+            "kv adapter doesn't support this operation",
+        )
+        .with_operation("kv::Adapter::append"))
+    }
+
+    /// Append a key into service
+    /// in blocking way.
+    fn blocking_append(&self, path: &str, value: &[u8]) -> Result<()> {
+        let _ = path;
+        let _ = value;
+
+        Err(Error::new(
+            ErrorKind::Unsupported,
+            "kv adapter doesn't support this operation",
+        )
+        .with_operation("kv::Adapter::blocking_append"))
+    }
 }
 
 /// Metadata for this key value accessor.

src/raw/adapters/kv/backend.rs

@@ -285,13 +285,20 @@ impl<S: Adapter> oio::Write for KvWriter<S> {
     }
 
     async fn append(&mut self, bs: Bytes) -> Result<()> {
-        self.buf.extend(bs);
-
+        if let Err(e) = self.kv.append(&self.path, bs.to_vec().as_slice()).await {
+            if e.kind() == ErrorKind::Unsupported {
+                self.buf.extend(bs);
+            } else {
+                return Err(e);
+            }
+        }
         Ok(())
     }
 
     async fn close(&mut self) -> Result<()> {
-        self.kv.set(&self.path, &self.buf).await?;
+        if !self.buf.is_empty() {
+            self.kv.set(&self.path, &self.buf).await?;
+        }
 
         Ok(())
     }

src/services/redis/backend.rs

@@ -340,6 +340,12 @@ impl kv::Adapter for Adapter {
         let _: () = conn.del(key).await?;
         Ok(())
     }
+
+    async fn append(&self, key: &str, value: &[u8]) -> Result<()> {
+        let mut conn = self.conn().await?;
+        conn.append(key, value).await?;
+        Ok(())
+    }
 }
 
 impl From<RedisError> for Error {