You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/09/11 17:00:39 UTC
(opendal) 01/01: fix(services/goosefs): return metadata from read streams
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/goosefs-read-metadata
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 75d4d51f97cb710fdc175c7c32d1b88177cc1253
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sat Sep 12 01:00:22 2026 +0800
fix(services/goosefs): return metadata from read streams
---
core/Cargo.lock | 1 +
core/services/goosefs/Cargo.toml | 1 +
core/services/goosefs/src/docs.md | 2 +
core/services/goosefs/src/reader.rs | 111 +++++++-----------------------------
core/services/goosefs/tests/read.rs | 69 ++++++++++++++++++++++
5 files changed, 95 insertions(+), 89 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index baed6ae12..fbb79942d 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -7039,6 +7039,7 @@ version = "0.59.1"
dependencies = [
"asyncband",
"bytes",
+ "futures",
"goosefs-sdk",
"log",
"opendal",
diff --git a/core/services/goosefs/Cargo.toml b/core/services/goosefs/Cargo.toml
index 1bab1d398..c6a2d007f 100644
--- a/core/services/goosefs/Cargo.toml
+++ b/core/services/goosefs/Cargo.toml
@@ -49,6 +49,7 @@ opendal-core = { path = "../../core", version = "0.59.1", default-features = fal
serde = { workspace = true, features = ["derive"] }
[dev-dependencies]
+futures = { workspace = true }
opendal = { path = "../..", version = "0.59.1", features = [
"services-goosefs",
] }
diff --git a/core/services/goosefs/src/docs.md b/core/services/goosefs/src/docs.md
index 3030b4e1b..c9a3f7a7f 100644
--- a/core/services/goosefs/src/docs.md
+++ b/core/services/goosefs/src/docs.md
@@ -23,6 +23,8 @@ which means it connects directly to GooseFS Master (port 9200) and
Worker (port 9203) without requiring a Proxy component.
Features:
+- **Read metadata**: Streams expose full-file metadata from the SDK reader before
+ consuming the body. Bounded reads do not issue a separate stat to obtain it.
- **HA support**: Comma-separated master addresses for automatic Primary Master discovery.
- **Block-level I/O**: Data reads/writes go through block-level gRPC bidirectional streaming.
- **Consistent hash routing**: Worker selection uses consistent hashing on block IDs.
diff --git a/core/services/goosefs/src/reader.rs b/core/services/goosefs/src/reader.rs
index cd746419d..4bbed3855 100644
--- a/core/services/goosefs/src/reader.rs
+++ b/core/services/goosefs/src/reader.rs
@@ -16,94 +16,25 @@
// under the License.
use super::backend::*;
-use super::core::GoosefsCore;
use super::core::{ErrorContext, parse_error};
use goosefs_sdk::io::GoosefsFileReader as SdkReader;
use opendal_core::raw::*;
use opendal_core::*;
-use std::sync::Arc;
pub struct GoosefsReadStream {
- core: Arc<GoosefsCore>,
- path: String,
- range: BytesRange,
- content_length: Option<u64>,
- /// Lazily-opened SDK reader. `None` until the first `read()` call.
- inner: Option<SdkReader>,
- /// Terminal flag: once the underlying stream has returned `None`,
- /// subsequent `read()` calls must keep returning an empty `Buffer`
- /// without re-entering the SDK (which would try to open a fresh
- /// stream and re-read the file).
+ inner: SdkReader,
+ // The SDK may reopen its stream after EOF; keep OpenDAL's EOF terminal.
done: bool,
}
-impl GoosefsReadStream {
- pub fn new(
- core: Arc<GoosefsCore>,
- path: String,
- range: BytesRange,
- content_length: Option<u64>,
- ) -> Self {
- GoosefsReadStream {
- core,
- path,
- range,
- content_length,
- inner: None,
- done: false,
- }
- }
-
- /// Open the underlying SDK reader, picking between the full-file
- /// and ranged variants based on `self.range`.
- async fn open(&self) -> Result<SdkReader> {
- let offset = self.range.offset();
- let size = self.range.size();
-
- // Three cases:
- // 1. No offset and no size → full-file stream.
- // 2. Offset+size both set → ranged stream.
- // 3. Offset set, size unbounded → resolve tail length via
- // get_status and fall through to the ranged opener, so we
- // still get per-block streaming (rather than buffering the
- // whole tail in one call).
- match (offset, size) {
- (0, None) => self.core.open_reader(&self.path).await,
- (off, Some(len)) => self.core.open_range_reader(&self.path, off, len).await,
- (off, None) => {
- let content_length = self.content_length.ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "content length must be known for offset reads",
- )
- })?;
- let len = content_length.saturating_sub(off);
- if len == 0 {
- // Empty tail — short-circuit with a zero-length
- // ranged open so the very next `read_next_block`
- // call returns `None` and we EOF cleanly.
- self.core.open_range_reader(&self.path, off, 0).await
- } else {
- self.core.open_range_reader(&self.path, off, len).await
- }
- }
- }
- }
-}
-
impl oio::ReadStream for GoosefsReadStream {
async fn read(&mut self) -> Result<Buffer> {
if self.done {
return Ok(Buffer::new());
}
- // Lazy open on first call.
- if self.inner.is_none() {
- self.inner = Some(self.open().await?);
- }
- let reader = self.inner.as_mut().expect("inner was just set");
-
- match reader
+ match self
+ .inner
.read_next_block()
.await
.map_err(|err| parse_error(ErrorContext::new(ServiceOperation("ReadNextBlock")), err))?
@@ -137,25 +68,27 @@ impl oio::StreamRead for GoosefsReader {
let backend = &self.backend;
let path = self.path.as_str();
- let content_length = if range.offset() != 0 && range.size().is_none() {
- let file_info = backend.core.get_status(path).await?;
- Some(
- backend
+ let inner = match (range.offset(), range.size()) {
+ (0, None) => backend.core.open_reader(path).await?,
+ (offset, Some(length)) => backend.core.open_range_reader(path, offset, length).await?,
+ (offset, None) => {
+ let file_info = backend.core.get_status(path).await?;
+ let content_length = backend
.core
.file_info_to_metadata(&file_info)?
- .content_length(),
- )
- } else {
- None
+ .content_length();
+ backend
+ .core
+ .open_range_reader(path, offset, content_length.saturating_sub(offset))
+ .await?
+ }
};
- let rp = RpRead::default();
- let stream = GoosefsReadStream::new(
- backend.core.clone(),
- path.to_string(),
- range,
- content_length,
- );
+ let metadata = backend.core.file_info_to_metadata(inner.file_info())?;
+ let stream = GoosefsReadStream { inner, done: false };
- Ok((rp, Box::new(stream) as Box<dyn oio::ReadStreamDyn>))
+ Ok((
+ RpRead::new(metadata),
+ Box::new(stream) as Box<dyn oio::ReadStreamDyn>,
+ ))
}
}
diff --git a/core/services/goosefs/tests/read.rs b/core/services/goosefs/tests/read.rs
new file mode 100644
index 000000000..f72ef6ecc
--- /dev/null
+++ b/core/services/goosefs/tests/read.rs
@@ -0,0 +1,69 @@
+// 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.
+
+//! Read metadata and stream contents against the GooseFS behavior-test fixture.
+
+use futures::TryStreamExt;
+use opendal::{BytesRange, Operator};
+
+#[tokio::test]
+async fn read_stream_metadata_describes_the_full_file() -> opendal::Result<()> {
+ let Ok(master) = std::env::var("OPENDAL_GOOSEFS_MASTER_ADDR") else {
+ eprintln!("skip: OPENDAL_GOOSEFS_MASTER_ADDR unset");
+ return Ok(());
+ };
+ let root = std::env::var("OPENDAL_GOOSEFS_ROOT").unwrap_or_else(|_| "/".to_string());
+ let op = Operator::new(
+ opendal::services::GooseFs::default()
+ .master_addr(&master)
+ .root(&root)
+ .write_type("must_cache"),
+ )?;
+ let path = format!("read-metadata-{}", std::process::id());
+ let data = b"0123456789abcdef";
+ op.write(&path, data.as_slice()).await?;
+ let expected = op.stat(&path).await?;
+
+ let result = async {
+ for (range, selected) in [
+ (BytesRange::from(..), 0..16),
+ (BytesRange::from(3..8), 3..8),
+ (BytesRange::from(3..), 3..16),
+ (BytesRange::suffix(5), 11..16),
+ ] {
+ let mut stream = op.reader(&path).await?.into_bytes_stream(range).await?;
+ let metadata = stream.metadata().await?;
+ assert_eq!(metadata.content_length(), data.len() as u64);
+ assert_eq!(metadata.etag(), expected.etag());
+ assert_eq!(metadata.last_modified(), expected.last_modified());
+
+ let mut actual = Vec::new();
+ while let Some(chunk) = stream.try_next().await.map_err(|error| {
+ opendal::Error::new(opendal::ErrorKind::Unexpected, "read stream failed")
+ .set_source(error)
+ })? {
+ actual.extend_from_slice(&chunk);
+ }
+ assert_eq!(actual, &data[selected]);
+ assert!(stream.try_next().await.unwrap().is_none());
+ }
+ Ok(())
+ }
+ .await;
+ op.delete(&path).await?;
+ result
+}