You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by er...@apache.org on 2025/06/19 19:58:50 UTC
(opendal) branch services-fs updated (e598ca276 -> 8cd5167ff)
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a change to branch services-fs
in repository https://gitbox.apache.org/repos/asf/opendal.git
discard e598ca276 refactor(services/fs): extract implementation to core
new 8cd5167ff refactor(services/fs): extract implementation to core
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (e598ca276)
\
N -- N -- N refs/heads/services-fs (8cd5167ff)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/src/services/fs/reader.rs | 36 +++++++++++++++++++++++++++++++++++-
1 file changed, 35 insertions(+), 1 deletion(-)
(opendal) 01/01: refactor(services/fs): extract implementation to core
Posted by er...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a commit to branch services-fs
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 8cd5167ff94ee837161c1d72f013f4e99d9a66b0
Author: Erick Guan <29...@users.noreply.github.com>
AuthorDate: Thu Jun 19 21:49:37 2025 +0200
refactor(services/fs): extract implementation to core
---
core/src/services/fs/backend.rs | 202 +++------------------------
core/src/services/fs/core.rs | 166 ++++++++++++++++++++++
core/src/services/fs/delete.rs | 6 +-
core/src/services/fs/{delete.rs => error.rs} | 51 +++----
core/src/services/fs/mod.rs | 2 +
5 files changed, 213 insertions(+), 214 deletions(-)
diff --git a/core/src/services/fs/backend.rs b/core/src/services/fs/backend.rs
index 593a00d25..a06ab0408 100644
--- a/core/src/services/fs/backend.rs
+++ b/core/src/services/fs/backend.rs
@@ -15,11 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use std::io::SeekFrom;
use std::path::PathBuf;
use std::sync::Arc;
-use chrono::DateTime;
use log::debug;
use super::core::*;
@@ -120,13 +118,7 @@ impl Builder for FsBuilder {
// Canonicalize the root directory. This should work since we already know that we can
// get the metadata of the path.
let root = root.canonicalize().map_err(|e| {
- Error::new(
- ErrorKind::Unexpected,
- "canonicalize of root directory failed",
- )
- .with_operation("Builder::build")
- .with_context("root", root.to_string_lossy())
- .set_source(e)
+ Error::new(ErrorKind::Unexpected,"canonicalize of root directory failed").set_source(e)
})?;
// Canonicalize the atomic_write_dir directory. This should work since we already know that
@@ -134,10 +126,7 @@ impl Builder for FsBuilder {
let atomic_write_dir = atomic_write_dir
.map(|p| {
p.canonicalize().map(Some).map_err(|e| {
- Error::new(
- ErrorKind::Unexpected,
- "canonicalize of atomic_write_dir directory failed",
- )
+ Error::new(ErrorKind::Unexpected, "canonicalize of atomic_write_dir directory failed")
.with_operation("Builder::build")
.with_context("root", root.to_string_lossy())
.set_source(e)
@@ -202,35 +191,12 @@ impl Access for FsBackend {
}
async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- tokio::fs::create_dir_all(&p)
- .await
- .map_err(new_std_io_error)?;
-
+ self.core.fs_create_dir(path).await?;
Ok(RpCreateDir::default())
}
async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let meta = tokio::fs::metadata(&p).await.map_err(new_std_io_error)?;
-
- let mode = if meta.is_dir() {
- EntryMode::DIR
- } else if meta.is_file() {
- EntryMode::FILE
- } else {
- EntryMode::Unknown
- };
- let m = Metadata::new(mode)
- .with_content_length(meta.len())
- .with_last_modified(
- meta.modified()
- .map(DateTime::from)
- .map_err(new_std_io_error)?,
- );
-
+ let m = self.core.fs_stat(path).await?;
Ok(RpStat::new(m))
}
@@ -244,22 +210,7 @@ impl Access for FsBackend {
///
/// Benchmark could be found [here](https://gist.github.com/Xuanwo/48f9cfbc3022ea5f865388bb62e1a70f)
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let mut f = tokio::fs::OpenOptions::new()
- .read(true)
- .open(&p)
- .await
- .map_err(new_std_io_error)?;
-
- if args.range().offset() != 0 {
- use tokio::io::AsyncSeekExt;
-
- f.seek(SeekFrom::Start(args.range().offset()))
- .await
- .map_err(new_std_io_error)?;
- }
-
+ let f = self.core.fs_read(path, &args).await?;
let r = FsReader::new(
self.core.clone(),
f,
@@ -269,80 +220,22 @@ impl Access for FsBackend {
}
async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let (target_path, tmp_path) = if let Some(atomic_write_dir) = &self.core.atomic_write_dir {
- let target_path = self
- .core
- .ensure_write_abs_path(&self.core.root, path)
- .await?;
- let tmp_path = self
- .core
- .ensure_write_abs_path(atomic_write_dir, &tmp_file_of(path))
- .await?;
-
- // If the target file exists, we should append to the end of it directly.
- if op.append()
- && tokio::fs::try_exists(&target_path)
- .await
- .map_err(new_std_io_error)?
- {
- (target_path, None)
- } else {
- (target_path, Some(tmp_path))
- }
- } else {
- let p = self
- .core
- .ensure_write_abs_path(&self.core.root, path)
- .await?;
-
- (p, None)
- };
-
- let mut open_options = tokio::fs::OpenOptions::new();
- if op.if_not_exists() {
- open_options.create_new(true);
- } else {
- open_options.create(true);
- }
+ let (target_path, tmp_path) = self.core.prepare_write(path, &op).await?;
+ let file = self.core.fs_write(&target_path, tmp_path.as_ref(), &op).await?;
- open_options.write(true);
+ let writer = FsWriter::new(target_path, tmp_path, file);
- if op.append() {
- open_options.append(true);
- } else {
- open_options.truncate(true);
- }
-
- let f = open_options
- .open(tmp_path.as_ref().unwrap_or(&target_path))
- .await
- .map_err(|e| {
- match e.kind() {
- std::io::ErrorKind::AlreadyExists => {
- // Map io AlreadyExists to opendal ConditionNotMatch
- Error::new(
- ErrorKind::ConditionNotMatch,
- "The file already exists in the filesystem",
- )
- .set_source(e)
- }
- _ => new_std_io_error(e),
- }
- })?;
-
- let w = FsWriter::new(target_path, tmp_path, f);
-
- let w = if op.append() {
- FsWriters::One(w)
+ let writer = if op.append() {
+ FsWriters::One(writer)
} else {
FsWriters::Two(oio::PositionWriter::new(
self.info().clone(),
- w,
+ writer,
op.concurrent(),
))
};
- Ok((RpWrite::default(), w))
+ Ok((RpWrite::default(), writer))
}
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
@@ -353,77 +246,22 @@ impl Access for FsBackend {
}
async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let f = match tokio::fs::read_dir(&p).await {
- Ok(rd) => rd,
- Err(e) => {
- return match e.kind() {
- // Return empty list if the directory not found
- std::io::ErrorKind::NotFound => Ok((RpList::default(), None)),
- // TODO: enable after our MSRV has been raised to 1.83
- //
- // If the path is not a directory, return an empty list
- //
- // The path could be a file or a symbolic link in this case.
- // Returning a NotADirectory error to the user isn't helpful; instead,
- // providing an empty directory is a more user-friendly. In fact, the dir
- // `path/` does not exist.
- // std::io::ErrorKind::NotADirectory => Ok((RpList::default(), None)),
- _ => {
- // TODO: remove this after we have MSRV 1.83
- #[cfg(unix)]
- if e.raw_os_error() == Some(20) {
- // On unix 20: Not a directory
- return Ok((RpList::default(), None));
- }
- #[cfg(windows)]
- if e.raw_os_error() == Some(267) {
- // On windows 267: DIRECTORY
- return Ok((RpList::default(), None));
- }
-
- Err(new_std_io_error(e))
- }
- };
+ match self.core.fs_list(path).await? {
+ Some(f) => {
+ let rd = FsLister::new(&self.core.root, path, f);
+ Ok((RpList::default(), Some(rd)))
}
- };
-
- let rd = FsLister::new(&self.core.root, path, f);
- Ok((RpList::default(), Some(rd)))
+ None => Ok((RpList::default(), None)),
+ }
}
async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let from = self.core.root.join(from.trim_end_matches('/'));
-
- // try to get the metadata of the source file to ensure it exists
- tokio::fs::metadata(&from).await.map_err(new_std_io_error)?;
-
- let to = self
- .core
- .ensure_write_abs_path(&self.core.root, to.trim_end_matches('/'))
- .await?;
-
- tokio::fs::copy(from, to).await.map_err(new_std_io_error)?;
-
+ self.core.fs_copy(from, to).await?;
Ok(RpCopy::default())
}
async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
- let from = self.core.root.join(from.trim_end_matches('/'));
-
- // try to get the metadata of the source file to ensure it exists
- tokio::fs::metadata(&from).await.map_err(new_std_io_error)?;
-
- let to = self
- .core
- .ensure_write_abs_path(&self.core.root, to.trim_end_matches('/'))
- .await?;
-
- tokio::fs::rename(from, to)
- .await
- .map_err(new_std_io_error)?;
-
+ self.core.fs_rename(from, to).await?;
Ok(RpRename::default())
}
}
diff --git a/core/src/services/fs/core.rs b/core/src/services/fs/core.rs
index 20a85e233..2fe67fa75 100644
--- a/core/src/services/fs/core.rs
+++ b/core/src/services/fs/core.rs
@@ -15,12 +15,15 @@
// specific language governing permissions and limitations
// under the License.
+use std::io::SeekFrom;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
+use chrono::DateTime;
use uuid::Uuid;
+use super::error::*;
use crate::raw::*;
use crate::*;
@@ -60,6 +63,169 @@ impl FsCore {
Ok(p)
}
+
+ pub async fn fs_create_dir(&self, path: &str) -> Result<()> {
+ let p = self.root.join(path.trim_end_matches('/'));
+ tokio::fs::create_dir_all(&p)
+ .await
+ .map_err(new_std_io_error)?;
+ Ok(())
+ }
+
+ pub async fn fs_stat(&self, path: &str) -> Result<Metadata> {
+ let p = self.root.join(path.trim_end_matches('/'));
+ let meta = tokio::fs::metadata(&p).await.map_err(new_std_io_error)?;
+
+ let mode = if meta.is_dir() {
+ EntryMode::DIR
+ } else if meta.is_file() {
+ EntryMode::FILE
+ } else {
+ EntryMode::Unknown
+ };
+ let m = Metadata::new(mode)
+ .with_content_length(meta.len())
+ .with_last_modified(
+ meta.modified()
+ .map(DateTime::from)
+ .map_err(new_std_io_error)?,
+ );
+
+ Ok(m)
+ }
+
+ pub async fn fs_read(&self, path: &str, args: &OpRead) -> Result<tokio::fs::File> {
+ let p = self.root.join(path.trim_end_matches('/'));
+
+ let mut f = tokio::fs::OpenOptions::new()
+ .read(true)
+ .open(&p)
+ .await
+ .map_err(new_std_io_error)?;
+
+ if args.range().offset() != 0 {
+ use tokio::io::AsyncSeekExt;
+ f.seek(SeekFrom::Start(args.range().offset()))
+ .await
+ .map_err(new_std_io_error)?;
+ }
+
+ Ok(f)
+ }
+
+ pub async fn prepare_write(&self, path: &str, op: &OpWrite) -> Result<(PathBuf, Option<PathBuf>)> {
+ let (target_path, tmp_path) = if let Some(atomic_write_dir) = &self.atomic_write_dir {
+ let target_path = self.ensure_write_abs_path(&self.root, path).await?;
+ let tmp_path = self
+ .ensure_write_abs_path(atomic_write_dir, &tmp_file_of(path))
+ .await?;
+
+ // If the target file exists, we should append to the end of it directly.
+ if op.append()
+ && tokio::fs::try_exists(&target_path)
+ .await
+ .map_err(new_std_io_error)?
+ {
+ (target_path, None)
+ } else {
+ (target_path, Some(tmp_path))
+ }
+ } else {
+ let p = self.ensure_write_abs_path(&self.root, path).await?;
+ (p, None)
+ };
+
+ Ok((target_path, tmp_path))
+ }
+
+ pub async fn fs_write(&self, target_path: &PathBuf, tmp_path: Option<&PathBuf>, op: &OpWrite) -> Result<tokio::fs::File> {
+ let mut open_options = tokio::fs::OpenOptions::new();
+ if op.if_not_exists() {
+ open_options.create_new(true);
+ } else {
+ open_options.create(true);
+ }
+
+ open_options.write(true);
+
+ if op.append() {
+ open_options.append(true);
+ } else {
+ open_options.truncate(true);
+ }
+
+ let f = open_options
+ .open(tmp_path.unwrap_or(target_path))
+ .await
+ .map_err(new_fs_io_error)?;
+
+ Ok(f)
+ }
+
+ pub async fn fs_list(&self, path: &str) -> Result<Option<tokio::fs::ReadDir>> {
+ let p = self.root.join(path.trim_end_matches('/'));
+
+ match tokio::fs::read_dir(&p).await {
+ Ok(rd) => Ok(Some(rd)),
+ Err(e) => {
+ match e.kind() {
+ // Return empty list if the directory not found
+ std::io::ErrorKind::NotFound => Ok(None),
+ // TODO: enable after our MSRV has been raised to 1.83
+ //
+ // If the path is not a directory, return an empty list
+ //
+ // The path could be a file or a symbolic link in this case.
+ // Returning a NotADirectory error to the user isn't helpful; instead,
+ // providing an empty directory is a more user-friendly. In fact, the dir
+ // `path/` does not exist.
+ // std::io::ErrorKind::NotADirectory => Ok((RpList::default(), None)),
+ _ => {
+ // TODO: remove this after we have MSRV 1.83
+ #[cfg(unix)]
+ if e.raw_os_error() == Some(20) {
+ // On unix 20: Not a directory
+ return Ok(None);
+ }
+ #[cfg(windows)]
+ if e.raw_os_error() == Some(267) {
+ // On windows 267: DIRECTORY
+ return Ok(None);
+ }
+
+ Err(new_std_io_error(e))
+ }
+ }
+ }
+ }
+ }
+
+ pub async fn fs_copy(&self, from: &str, to: &str) -> Result<()> {
+ let from = self.root.join(from.trim_end_matches('/'));
+ // try to get the metadata of the source file to ensure it exists
+ tokio::fs::metadata(&from).await.map_err(new_std_io_error)?;
+
+ let to = self
+ .ensure_write_abs_path(&self.root, to.trim_end_matches('/'))
+ .await?;
+
+ tokio::fs::copy(from, to).await.map_err(new_std_io_error)?;
+ Ok(())
+ }
+
+ pub async fn fs_rename(&self, from: &str, to: &str) -> Result<()> {
+ let from = self.root.join(from.trim_end_matches('/'));
+ tokio::fs::metadata(&from).await.map_err(new_std_io_error)?;
+
+ let to = self
+ .ensure_write_abs_path(&self.root, to.trim_end_matches('/'))
+ .await?;
+
+ tokio::fs::rename(from, to)
+ .await
+ .map_err(new_std_io_error)?;
+ Ok(())
+ }
}
#[inline]
diff --git a/core/src/services/fs/delete.rs b/core/src/services/fs/delete.rs
index f14e6d13e..40040d984 100644
--- a/core/src/services/fs/delete.rs
+++ b/core/src/services/fs/delete.rs
@@ -18,6 +18,7 @@
use std::sync::Arc;
use super::core::*;
+use super::error::*;
use crate::raw::*;
use crate::*;
@@ -40,11 +41,10 @@ impl oio::OneShotDelete for FsDeleter {
match meta {
Ok(meta) => {
if meta.is_dir() {
- tokio::fs::remove_dir(&p).await.map_err(new_std_io_error)?;
+ ignore_not_found(tokio::fs::remove_dir(&p).await, ())?;
} else {
- tokio::fs::remove_file(&p).await.map_err(new_std_io_error)?;
+ ignore_not_found(tokio::fs::remove_file(&p).await, ())?;
}
-
Ok(())
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
diff --git a/core/src/services/fs/delete.rs b/core/src/services/fs/error.rs
similarity index 50%
copy from core/src/services/fs/delete.rs
copy to core/src/services/fs/error.rs
index f14e6d13e..08f804b80 100644
--- a/core/src/services/fs/delete.rs
+++ b/core/src/services/fs/error.rs
@@ -15,40 +15,33 @@
// specific language governing permissions and limitations
// under the License.
-use std::sync::Arc;
-
-use super::core::*;
use crate::raw::*;
use crate::*;
-pub struct FsDeleter {
- core: Arc<FsCore>,
-}
-
-impl FsDeleter {
- pub fn new(core: Arc<FsCore>) -> Self {
- Self { core }
+#[cfg(unix)]
+const UNIX_NOT_A_DIRECTORY: i32 = 20;
+#[cfg(windows)]
+const WINDOWS_NOT_A_DIRECTORY: i32 = 267;
+
+/// Enhanced filesystem-specific I/O error conversion with semantic mappings
+pub fn new_fs_io_error(e: std::io::Error) -> Error {
+ match e.kind() {
+ std::io::ErrorKind::AlreadyExists => {
+ Error::new(
+ ErrorKind::ConditionNotMatch,
+ "The file already exists in the filesystem",
+ )
+ .set_source(e)
+ }
+ _ => new_std_io_error(e),
}
}
-impl oio::OneShotDelete for FsDeleter {
- async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let p = self.core.root.join(path.trim_end_matches('/'));
-
- let meta = tokio::fs::metadata(&p).await;
-
- match meta {
- Ok(meta) => {
- if meta.is_dir() {
- tokio::fs::remove_dir(&p).await.map_err(new_std_io_error)?;
- } else {
- tokio::fs::remove_file(&p).await.map_err(new_std_io_error)?;
- }
-
- Ok(())
- }
- Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
- Err(err) => Err(new_std_io_error(err)),
- }
+/// Convert "not found" errors to success for idempotent operations like delete
+pub fn ignore_not_found<T>(result: std::io::Result<T>, default: T) -> Result<T> {
+ match result {
+ Ok(value) => Ok(value),
+ Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(default),
+ Err(e) => Err(new_std_io_error(e)),
}
}
diff --git a/core/src/services/fs/mod.rs b/core/src/services/fs/mod.rs
index caf858386..cbfefd3cf 100644
--- a/core/src/services/fs/mod.rs
+++ b/core/src/services/fs/mod.rs
@@ -20,6 +20,8 @@ mod core;
#[cfg(feature = "services-fs")]
mod delete;
#[cfg(feature = "services-fs")]
+mod error;
+#[cfg(feature = "services-fs")]
mod lister;
#[cfg(feature = "services-fs")]
mod reader;