feat(service/fs): add append support for fs

core/src/services/fs/appender.rs

@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use async_trait::async_trait;
+use bytes::Bytes;
+
+use tokio::io::AsyncWriteExt;
+
+use super::error::parse_io_error;
+use crate::raw::*;
+use crate::*;
+
+pub struct FsAppender<F> {
+ f: F,
+}
+
+impl<F> FsAppender<F> {
+ pub fn new(f: F) -> Self {
+ Self { f }
+ }
+}
+
+#[async_trait]
+impl oio::Append for FsAppender<tokio::fs::File> {
+ async fn append(&mut self, bs: Bytes) -> Result<()> {
+ self.f.write_all(&bs).await.map_err(parse_io_error)?;
+
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<()> {
+ self.f.sync_all().await.map_err(parse_io_error)?;
+
+ Ok(())
+ }
+}

core/src/services/fs/backend.rs

@@ -32,6 +32,7 @@ use super::pager::FsPager;
 use super::writer::FsWriter;
 use crate::ops::*;
 use crate::raw::*;
+use crate::services::fs::appender::FsAppender;
 use crate::*;
 
 /// POSIX file system support.
@@ -43,6 +44,7 @@ use crate::*;
 /// - [x] stat
 /// - [x] read
 /// - [x] write
+/// - [x] append
 /// - [x] create_dir
 /// - [x] delete
 /// - [x] copy
@@ -296,7 +298,7 @@ impl Accessor for FsBackend {
  type BlockingReader = oio::into_blocking_reader::FdReader<std::fs::File>;
  type Writer = FsWriter<tokio::fs::File>;
  type BlockingWriter = FsWriter<std::fs::File>;
- type Appender = ();
+ type Appender = FsAppender<tokio::fs::File>;
  type Pager = Option<FsPager<tokio::fs::ReadDir>>;
  type BlockingPager = Option<FsPager<std::fs::ReadDir>>;
 
@@ -316,6 +318,8 @@ impl Accessor for FsBackend {
  create_dir: true,
  delete: true,
 
+ append: true,
+
  list: true,
  list_with_delimiter_slash: true,
 
@@ -434,6 +438,20 @@ impl Accessor for FsBackend {
  Ok((RpWrite::new(), FsWriter::new(target_path, tmp_path, f)))
  }
 
+ async fn append(&self, path: &str, _: OpAppend) -> Result<(RpAppend, Self::Appender)> {
+ let path = Self::ensure_write_abs_path(&self.root, path).await?;
+
+ let f = tokio::fs::OpenOptions::new()
+ .create(true)
+ .write(true)
+ .append(true)
+ .open(&path)
+ .await
+ .map_err(parse_io_error)?;
+
+ Ok((RpAppend::new(), FsAppender::new(f)))
+ }
+
  async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
  let from = self.root.join(from.trim_end_matches('/'));
 

core/src/services/fs/mod.rs

@@ -18,6 +18,7 @@
 mod backend;
 pub use backend::FsBuilder as Fs;
 
+mod appender;
 mod error;
 mod pager;
 mod writer;