This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new e002dccfd feat(services/hf): add download_mode option and require
explicit repo_type (#7625)
e002dccfd is described below
commit e002dccfd4bde2fb51e6fa9d60256b48531e2966
Author: Krisztián Szűcs <[email protected]>
AuthorDate: Thu May 28 11:02:42 2026 +0200
feat(services/hf): add download_mode option and require explicit repo_type
(#7625)
* feat(services/hf): add download_mode option and require explicit repo_type
* test(services/hf): add network test for non-XET file HTTP fallback in Xet
mode
* chore: update Cargo.lock for base64 dev-dependency in hf service
* fix(services/hf): move config enums out of config
* fix(services/hf): return metadata from read
* test(services/hf): add bucket http behavior setup
---------
Co-authored-by: Xuanwo <[email protected]>
---
.github/services/hf/hf_bucket_http/action.yml | 42 +++++
.../java/org/apache/opendal/ServiceConfig.java | 16 +-
bindings/python/src/services.rs | 22 ++-
core/Cargo.lock | 1 +
core/services/hf/Cargo.toml | 1 +
core/services/hf/src/backend.rs | 99 +++++++---
core/services/hf/src/config.rs | 79 ++++++--
core/services/hf/src/core.rs | 206 ++++++++++++++-------
core/services/hf/src/error.rs | 24 +--
core/services/hf/src/reader.rs | 167 ++++++++++-------
core/services/hf/src/uri.rs | 60 +-----
11 files changed, 470 insertions(+), 247 deletions(-)
diff --git a/.github/services/hf/hf_bucket_http/action.yml
b/.github/services/hf/hf_bucket_http/action.yml
new file mode 100644
index 000000000..235b4726c
--- /dev/null
+++ b/.github/services/hf/hf_bucket_http/action.yml
@@ -0,0 +1,42 @@
+# 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.
+
+name: hf_bucket_http
+description: "Behavior test for Huggingface Bucket using HTTP downloads"
+
+runs:
+ using: "composite"
+ steps:
+ - name: Setup
+ uses:
1Password/load-secrets-action@dafbe7cb03502b260e2b2893c753c352eee545bf # v3.2.1
+ with:
+ export-env: true
+ env:
+ OPENDAL_HF_TOKEN: op://services/hf/token
+ - name: Create temp bucket
+ uses: ./.github/actions/hf-temp-repo
+ with:
+ repo_id: opendal/test-bucket-http-${{ github.run_id }}-${{ github.job
}}
+ repo_type: bucket
+ token: ${{ env.OPENDAL_HF_TOKEN }}
+ - name: Setup env
+ shell: bash
+ run: |
+ cat << EOF >> $GITHUB_ENV
+ OPENDAL_HF_REPO_TYPE=bucket
+ OPENDAL_HF_DOWNLOAD_MODE=http
+ EOF
diff --git a/bindings/java/src/main/java/org/apache/opendal/ServiceConfig.java
b/bindings/java/src/main/java/org/apache/opendal/ServiceConfig.java
index d38a17f31..d4cc5ad99 100644
--- a/bindings/java/src/main/java/org/apache/opendal/ServiceConfig.java
+++ b/bindings/java/src/main/java/org/apache/opendal/ServiceConfig.java
@@ -1573,6 +1573,10 @@ public interface ServiceConfig {
@Data
@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
class Hf implements ServiceConfig {
+ /**
+ * <p>Download mode. Either <code>xet</code> (default) or
<code>http</code>.</p>
+ */
+ public final String downloadMode;
/**
* <p>Endpoint of the Hugging Face Hub.</p>
* <p>Default is "https://huggingface.co".</p>
@@ -1584,10 +1588,9 @@ public interface ServiceConfig {
*/
public final String repoId;
/**
- * <p>Repo type of this backend. Default is model.</p>
- * <p>Default is model</p>
+ * <p>Repo type of this backend. Required.</p>
*/
- public final @NonNull String repoType;
+ public final String repoType;
/**
* <p>Revision of this backend.</p>
* <p>Default is main.</p>
@@ -1612,13 +1615,18 @@ public interface ServiceConfig {
@Override
public Map<String, String> configMap() {
final HashMap<String, String> map = new HashMap<>();
+ if (downloadMode != null) {
+ map.put("download_mode", downloadMode);
+ }
if (endpoint != null) {
map.put("endpoint", endpoint);
}
if (repoId != null) {
map.put("repo_id", repoId);
}
- map.put("repo_type", repoType);
+ if (repoType != null) {
+ map.put("repo_type", repoType);
+ }
if (revision != null) {
map.put("revision", revision);
}
diff --git a/bindings/python/src/services.rs b/bindings/python/src/services.rs
index 1db291955..0a00c460e 100644
--- a/bindings/python/src/services.rs
+++ b/bindings/python/src/services.rs
@@ -1128,9 +1128,10 @@ submit! {
scheme: typing.Literal[opendal.services.Scheme.Hf, "hf"],
/,
*,
+ download_mode: builtins.str = ...,
endpoint: builtins.str = ...,
repo_id: builtins.str = ...,
- repo_type: builtins.str,
+ repo_type: builtins.str = ...,
revision: builtins.str = ...,
root: builtins.str = ...,
token: builtins.str = ...,
@@ -1140,16 +1141,18 @@ submit! {
Parameters
----------
+ download_mode : builtins.str, optional
+ Download mode.
+ Either `xet` (default) or `http`.
endpoint : builtins.str, optional
Endpoint of the Hugging Face Hub.
Default is "https://huggingface.co";.
repo_id : builtins.str, optional
Repo id of this backend.
This is required.
- repo_type : builtins.str
+ repo_type : builtins.str, optional
Repo type of this backend.
- Default is model.
- Default is model
+ Required.
revision : builtins.str, optional
Revision of this backend.
Default is main.
@@ -3769,9 +3772,10 @@ submit! {
scheme: typing.Literal[opendal.services.Scheme.Hf, "hf"],
/,
*,
+ download_mode: builtins.str = ...,
endpoint: builtins.str = ...,
repo_id: builtins.str = ...,
- repo_type: builtins.str,
+ repo_type: builtins.str = ...,
revision: builtins.str = ...,
root: builtins.str = ...,
token: builtins.str = ...,
@@ -3781,16 +3785,18 @@ submit! {
Parameters
----------
+ download_mode : builtins.str, optional
+ Download mode.
+ Either `xet` (default) or `http`.
endpoint : builtins.str, optional
Endpoint of the Hugging Face Hub.
Default is "https://huggingface.co";.
repo_id : builtins.str, optional
Repo id of this backend.
This is required.
- repo_type : builtins.str
+ repo_type : builtins.str, optional
Repo type of this backend.
- Default is model.
- Default is model
+ Required.
revision : builtins.str, optional
Revision of this backend.
Default is main.
diff --git a/core/Cargo.lock b/core/Cargo.lock
index c5e953b13..f2d0b3de1 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -7057,6 +7057,7 @@ dependencies = [
name = "opendal-service-hf"
version = "0.57.0"
dependencies = [
+ "base64 0.22.1",
"bytes",
"futures",
"hf-xet",
diff --git a/core/services/hf/Cargo.toml b/core/services/hf/Cargo.toml
index 980bca1ce..af9af7e27 100644
--- a/core/services/hf/Cargo.toml
+++ b/core/services/hf/Cargo.toml
@@ -44,6 +44,7 @@ serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
[dev-dependencies]
+base64 = { workspace = true }
futures = { workspace = true }
opendal-core = { path = "../../core", version = "0.57.0", features = [
"reqwest-rustls-tls",
diff --git a/core/services/hf/src/backend.rs b/core/services/hf/src/backend.rs
index 2b7f860ba..dc6723f55 100644
--- a/core/services/hf/src/backend.rs
+++ b/core/services/hf/src/backend.rs
@@ -23,6 +23,7 @@ use log::debug;
use super::HF_SCHEME;
use super::config::HfConfig;
use super::core::HfCore;
+use super::core::HfDownloadMode;
use super::deleter::HfDeleter;
use super::lister::HfLister;
use super::reader::HfReader;
@@ -52,7 +53,7 @@ impl HfBuilder {
pub fn repo_type(mut self, repo_type: &str) -> Self {
if !repo_type.is_empty() {
if let Ok(rt) = HfRepoType::parse(repo_type) {
- self.config.repo_type = rt;
+ self.config.repo_type = Some(rt);
}
}
self
@@ -111,6 +112,19 @@ impl HfBuilder {
self
}
+ /// Set the download mode. Either `xet` (default) or `http`.
+ ///
+ /// - `xet`: uses the XET protocol for downloads (default).
+ /// - `http`: plain HTTP download, following the redirect from the server.
+ pub fn download_mode(mut self, mode: &str) -> Self {
+ if !mode.is_empty() {
+ if let Ok(m) = HfDownloadMode::parse(mode) {
+ self.config.download_mode = Some(m);
+ }
+ }
+ self
+ }
+
/// configure the Hub base url. You might want to set this variable if your
/// organization is using a Private Hub https://huggingface.co/enterprise
///
@@ -178,15 +192,18 @@ impl Builder for HfBuilder {
let token = self.hf_token();
let endpoint = self.hf_endpoint();
- let repo_type = self.config.repo_type;
+ let repo_type = self.config.repo_type.ok_or_else(|| {
+ Error::new(ErrorKind::ConfigInvalid, "repo_type is required")
+ .with_operation("Builder::build")
+ .with_context("service", HF_SCHEME)
+ })?;
debug!("backend use repo_type: {:?}", &repo_type);
- let repo_id = match &self.config.repo_id {
- Some(repo_id) => Ok(repo_id.clone()),
- None => Err(Error::new(ErrorKind::ConfigInvalid, "repo_id is
empty")
+ let repo_id = self.config.repo_id.ok_or_else(|| {
+ Error::new(ErrorKind::ConfigInvalid, "repo_id is required")
.with_operation("Builder::build")
- .with_context("service", HF_SCHEME)),
- }?;
+ .with_context("service", HF_SCHEME)
+ })?;
debug!("backend use repo_id: {}", &repo_id);
let revision = match &self.config.revision {
@@ -199,6 +216,8 @@ impl Builder for HfBuilder {
debug!("backend use root: {}", &root);
debug!("backend use token: {}", token.is_some());
debug!("backend use endpoint: {}", &endpoint);
+ let download_mode = self.config.download_mode.unwrap_or_default();
+ debug!("backend use download_mode: {:?}", download_mode);
let info: Arc<AccessorInfo> = {
let am = AccessorInfo::default();
@@ -220,7 +239,14 @@ impl Builder for HfBuilder {
debug!("backend repo uri: {:?}", repo.uri(&root, ""));
Ok(HfBackend {
- core: Arc::new(HfCore::build(info, repo, root, token, endpoint)?),
+ core: Arc::new(HfCore::build(
+ info,
+ repo,
+ root,
+ token,
+ endpoint,
+ download_mode,
+ )?),
})
}
}
@@ -248,19 +274,9 @@ impl Access for HfBackend {
return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
}
- if self.core.repo.is_bucket() {
- if path.ends_with('/') {
- return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
- }
- return match self.core.maybe_xet_file(path).await? {
- Some(file_info) => {
- let size = file_info.file_size().unwrap_or(0);
- Ok(RpStat::new(
-
Metadata::new(EntryMode::FILE).with_content_length(size),
- ))
- }
- None => Err(Error::new(ErrorKind::NotFound, "path not found")),
- };
+ // Buckets have no git directory entries; treat any trailing-slash
path as a virtual dir.
+ if self.core.repo.is_bucket() && path.ends_with('/') {
+ return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
}
let info = self.core.path_info(path).await?;
@@ -268,8 +284,7 @@ impl Access for HfBackend {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead,
Self::Reader)> {
- let (metadata, reader) = HfReader::try_new(&self.core, path,
args.range()).await?;
- Ok((RpRead::new(metadata), reader))
+ HfReader::try_new(&self.core, path, args.range()).await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList,
Self::Lister)> {
@@ -294,10 +309,15 @@ impl Access for HfBackend {
#[cfg(test)]
pub(super) mod test_utils {
+ use std::sync::Arc;
+
+ use super::super::core::{HfCore, HfDownloadMode};
+ use super::super::uri::{HfRepo, HfRepoType};
use super::HfBuilder;
+ use opendal_core::Capability;
use opendal_core::Operator;
use opendal_core::layers::HttpClientLayer;
- use opendal_core::raw::HttpClient;
+ use opendal_core::raw::{AccessorInfo, HttpClient};
fn finish_operator(op: Operator) -> Operator {
let client = HttpClient::with(reqwest::Client::new());
@@ -308,7 +328,8 @@ pub(super) mod test_utils {
let op = Operator::new(
HfBuilder::default()
.repo_type("model")
- .repo_id("openai-community/gpt2"),
+ .repo_id("openai-community/gpt2")
+ .download_mode("http"),
)
.unwrap()
.finish();
@@ -326,6 +347,34 @@ pub(super) mod test_utils {
finish_operator(op)
}
+ pub fn testing_dataset_core() -> Arc<HfCore> {
+ let repo_id =
std::env::var("HF_OPENDAL_DATASET").expect("HF_OPENDAL_DATASET must be set");
+ let token = std::env::var("HF_OPENDAL_TOKEN").expect("HF_OPENDAL_TOKEN
must be set");
+
+ let info = AccessorInfo::default();
+ info.set_scheme("hf").set_native_capability(Capability {
+ read: true,
+ write: true,
+ delete: true,
+ ..Default::default()
+ });
+ info.update_http_client(|_| HttpClient::with(reqwest::Client::new()));
+
+ let repo = HfRepo::new(HfRepoType::Dataset, repo_id,
Some("main".to_string()));
+
+ Arc::new(
+ HfCore::build(
+ Arc::new(info),
+ repo,
+ "/".to_string(),
+ Some(token),
+ "https://huggingface.co".to_string(),
+ HfDownloadMode::Xet,
+ )
+ .expect("failed to build HfCore"),
+ )
+ }
+
pub fn testing_bucket_operator() -> Operator {
let repo_id =
std::env::var("HF_OPENDAL_BUCKET").expect("HF_OPENDAL_BUCKET must be set");
let token = std::env::var("HF_OPENDAL_TOKEN").expect("HF_OPENDAL_TOKEN
must be set");
diff --git a/core/services/hf/src/config.rs b/core/services/hf/src/config.rs
index 808a88458..2e7094ee3 100644
--- a/core/services/hf/src/config.rs
+++ b/core/services/hf/src/config.rs
@@ -16,21 +16,22 @@
// under the License.
use super::backend::HfBuilder;
-use super::uri::HfRepoType;
+use super::core::HfDownloadMode;
+use super::core::HfRepoType;
use super::uri::HfUri;
use serde::Deserialize;
use serde::Serialize;
use std::fmt::Debug;
+use super::HUGGINGFACE_SCHEME;
+
/// Configuration for Hugging Face service support.
#[derive(Default, Serialize, Deserialize, Clone, PartialEq, Eq)]
#[serde(default)]
#[non_exhaustive]
pub struct HfConfig {
- /// Repo type of this backend. Default is model.
- ///
- /// Default is model
- pub repo_type: HfRepoType,
+ /// Repo type of this backend. Required.
+ pub repo_type: Option<HfRepoType>,
/// Repo id of this backend.
///
/// This is required.
@@ -51,15 +52,21 @@ pub struct HfConfig {
///
/// Default is "https://huggingface.co";.
pub endpoint: Option<String>,
+ /// Download mode. Either `xet` (default) or `http`.
+ pub download_mode: Option<HfDownloadMode>,
}
impl Debug for HfConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("HfConfig")
- .field("repo_type", &self.repo_type)
+ .field(
+ "repo_type",
+ &self.repo_type.as_ref().map(HfRepoType::as_str),
+ )
.field("repo_id", &self.repo_id)
.field("revision", &self.revision)
.field("root", &self.root)
+ .field("download_mode", &self.download_mode)
.finish_non_exhaustive()
}
}
@@ -88,32 +95,43 @@ impl opendal_core::Configurator for HfConfig {
}
}
+ let download_mode = opts
+ .get("download_mode")
+ .map(|s| HfDownloadMode::parse(s))
+ .transpose()?;
+
if !path.is_empty() {
// Full URI like "hf://datasets/user/repo@rev/path"
let parsed = HfUri::parse(&path)?;
Ok(Self {
- repo_type: parsed.repo.repo_type,
+ repo_type: Some(parsed.repo.repo_type),
repo_id: Some(parsed.repo.repo_id),
revision: parsed.repo.revision,
+ root: opts.get("root").cloned(),
token: opts.get("token").cloned(),
endpoint: opts.get("endpoint").cloned(),
- ..Default::default()
+ download_mode,
})
} else {
// Bare scheme from via_iter, all config is in options.
let repo_type = opts
.get("repo_type")
- .map(|s| HfRepoType::parse(s))
- .transpose()?
- .unwrap_or_default();
+ .ok_or_else(|| {
+ opendal_core::Error::new(
+ opendal_core::ErrorKind::ConfigInvalid,
+ "repo_type is required",
+ )
+ .with_context("service", HUGGINGFACE_SCHEME)
+ })
+ .and_then(|s| HfRepoType::parse(s))?;
Ok(Self {
- repo_type,
+ repo_type: Some(repo_type),
repo_id: opts.get("repo_id").cloned(),
revision: opts.get("revision").cloned(),
root: opts.get("root").cloned(),
token: opts.get("token").cloned(),
endpoint: opts.get("endpoint").cloned(),
- ..Default::default()
+ download_mode,
})
}
}
@@ -138,7 +156,7 @@ mod tests {
.unwrap();
let cfg = HfConfig::from_uri(&uri).unwrap();
- assert_eq!(cfg.repo_type, HfRepoType::Dataset);
+ assert_eq!(cfg.repo_type, Some(HfRepoType::Dataset));
assert_eq!(cfg.repo_id.as_deref(), Some("username/my_dataset"));
assert_eq!(cfg.revision.as_deref(), Some("dev"));
assert!(cfg.root.is_none());
@@ -162,9 +180,40 @@ mod tests {
.unwrap();
let cfg = HfConfig::from_uri(&uri).unwrap();
- assert_eq!(cfg.repo_type, HfRepoType::Dataset);
+ assert_eq!(cfg.repo_type, Some(HfRepoType::Dataset));
assert_eq!(cfg.repo_id.as_deref(),
Some("opendal/huggingface-testdata"));
assert_eq!(cfg.revision.as_deref(), Some("main"));
assert_eq!(cfg.root.as_deref(), Some("/testdata/"));
}
+
+ #[test]
+ fn from_uri_download_mode_http() {
+ let uri = OperatorUri::new(
+ "huggingface",
+ vec![
+ ("repo_type".to_string(), "dataset".to_string()),
+ ("repo_id".to_string(), "user/repo".to_string()),
+ ("download_mode".to_string(), "http".to_string()),
+ ],
+ )
+ .unwrap();
+
+ let cfg = HfConfig::from_uri(&uri).unwrap();
+ assert_eq!(cfg.download_mode, Some(HfDownloadMode::Http));
+ }
+
+ #[test]
+ fn from_uri_download_mode_defaults_to_xet() {
+ let uri = OperatorUri::new(
+ "huggingface",
+ vec![
+ ("repo_type".to_string(), "model".to_string()),
+ ("repo_id".to_string(), "user/repo".to_string()),
+ ],
+ )
+ .unwrap();
+
+ let cfg = HfConfig::from_uri(&uri).unwrap();
+ assert_eq!(cfg.download_mode.unwrap_or_default(), HfDownloadMode::Xet);
+ }
}
diff --git a/core/services/hf/src/core.rs b/core/services/hf/src/core.rs
index 62ffe1577..bc604f241 100644
--- a/core/services/hf/src/core.rs
+++ b/core/services/hf/src/core.rs
@@ -23,17 +23,89 @@ use bytes::Bytes;
use http::Request;
use http::Response;
use http::header;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
-use xet::xet_session::{
- XetDownloadStreamGroup, XetFileInfo, XetSession, XetSessionBuilder,
XetUploadCommit,
-};
+use xet::xet_session::{XetDownloadStreamGroup, XetSession, XetSessionBuilder,
XetUploadCommit};
-use super::error::parse_error;
-use super::uri::{HfRepo, HfUri};
use opendal_core::raw::*;
use opendal_core::*;
+use super::HUGGINGFACE_SCHEME;
+use super::error::parse_error;
+use super::uri::{HfRepo, HfUri};
+
+/// Repository type of Huggingface. Supports `model`, `dataset`, `space`, and
`bucket`.
+/// [Reference](https://huggingface.co/docs/hub/repositories)
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum HfRepoType {
+ Model,
+ Dataset,
+ Space,
+ Bucket,
+}
+
+impl HfRepoType {
+ pub fn parse(s: &str) -> Result<Self> {
+ match s.to_lowercase().replace(' ', "").as_str() {
+ "model" | "models" => Ok(Self::Model),
+ "dataset" | "datasets" => Ok(Self::Dataset),
+ "space" | "spaces" => Ok(Self::Space),
+ "bucket" | "buckets" => Ok(Self::Bucket),
+ other => Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ format!("unknown repo type: {other}"),
+ )
+ .with_context("service", HUGGINGFACE_SCHEME)),
+ }
+ }
+
+ pub fn as_str(&self) -> &'static str {
+ match self {
+ Self::Model => "model",
+ Self::Dataset => "dataset",
+ Self::Space => "space",
+ Self::Bucket => "bucket",
+ }
+ }
+
+ pub fn as_plural_str(&self) -> &'static str {
+ match self {
+ Self::Model => "models",
+ Self::Dataset => "datasets",
+ Self::Space => "spaces",
+ Self::Bucket => "buckets",
+ }
+ }
+}
+
+/// Download mode for HuggingFace files.
+///
+/// - `xet` (default): uses the XET protocol, asks resolve for XET file
metadata,
+/// and routes XET files through the CAS download stream.
+/// - `http`: follows the resolve redirect and streams bytes directly.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
+#[serde(rename_all = "lowercase")]
+pub enum HfDownloadMode {
+ #[default]
+ Xet,
+ Http,
+}
+
+impl HfDownloadMode {
+ pub fn parse(s: &str) -> Result<Self> {
+ match s.to_lowercase().as_str() {
+ "xet" => Ok(Self::Xet),
+ "http" => Ok(Self::Http),
+ other => Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ format!("unknown download mode: {other}"),
+ )
+ .with_context("service", HUGGINGFACE_SCHEME)),
+ }
+ }
+}
+
/// API payload structures for commit operations
#[derive(Debug, serde::Serialize)]
pub(super) struct CommitFile {
@@ -86,6 +158,12 @@ pub(super) struct MixedCommitPayload {
// API response types
+#[derive(Deserialize)]
+pub(super) struct XetFileResponse {
+ pub hash: String,
+ pub size: u64,
+}
+
#[derive(serde::Deserialize, Debug)]
pub(super) struct CommitResponse {
#[allow(dead_code)]
@@ -110,6 +188,9 @@ pub(super) struct PathInfo {
pub path: String,
#[serde(default)]
pub last_commit: Option<LastCommit>,
+ /// BLAKE3 Merkle hash for XET-stored files; absent for plain git or
non-XET LFS files.
+ #[serde(rename = "xetHash", default)]
+ pub xet_hash: Option<String>,
}
impl PathInfo {
@@ -150,6 +231,7 @@ pub(super) struct LfsInfo {
#[derive(Deserialize, Eq, PartialEq, Debug)]
pub(super) struct LastCommit {
+ pub id: String,
pub date: String,
}
@@ -163,10 +245,8 @@ pub struct HfCore {
pub root: String,
pub token: Option<String>,
pub endpoint: String,
- /// HTTP client with redirects disabled, used by XET probes to
- /// inspect headers on 302 responses.
- pub no_redirect_client: HttpClient,
pub xet_session: XetSession,
+ pub download_mode: HfDownloadMode,
}
impl Debug for HfCore {
@@ -186,8 +266,8 @@ impl HfCore {
root: String,
token: Option<String>,
endpoint: String,
- no_redirect_client: HttpClient,
xet_session: XetSession,
+ download_mode: HfDownloadMode,
) -> Self {
Self {
info,
@@ -195,28 +275,19 @@ impl HfCore {
root,
token,
endpoint,
- no_redirect_client,
xet_session,
+ download_mode,
}
}
- /// Build HfCore with dedicated reqwest HTTP clients.
- ///
- /// Uses separate clients for standard and no-redirect requests to
- /// avoid "dispatch task is gone" errors with multiple tokio runtimes.
pub fn build(
info: Arc<AccessorInfo>,
repo: HfRepo,
root: String,
token: Option<String>,
endpoint: String,
+ download_mode: HfDownloadMode,
) -> Result<Self> {
- let standard_client =
-
HttpClient::with(build_reqwest(reqwest::redirect::Policy::default())?);
- let no_redirect_client =
-
HttpClient::with(build_reqwest(reqwest::redirect::Policy::none())?);
- info.update_http_client(|_| standard_client);
-
let xet_session = XetSessionBuilder::new().build().map_err(|err| {
Error::new(ErrorKind::Unexpected, "failed to create xet
session").set_source(err)
})?;
@@ -227,8 +298,8 @@ impl HfCore {
root,
token,
endpoint,
- no_redirect_client,
xet_session,
+ download_mode,
))
}
@@ -337,7 +408,8 @@ impl HfCore {
if resp.status().is_success() {
Ok(resp)
} else {
- Err(parse_error(resp))
+ let (parts, body) = resp.into_parts();
+ Err(parse_error(parts, body))
}
}
@@ -374,46 +446,40 @@ impl HfCore {
Ok(files.remove(0))
}
- /// Issue a HEAD request and extract XET file info (hash and size).
+ /// Send `GET /resolve` and return the raw streaming response.
///
- /// Returns `None` if the `X-Xet-Hash` header is absent or empty.
- ///
- /// Uses a dedicated no-redirect HTTP client so we can inspect
- /// headers (e.g. `X-Xet-Hash`) on the 302 response.
- pub(super) async fn maybe_xet_file(&self, path: &str) ->
Result<Option<XetFileInfo>> {
+ /// In `Xet` mode adds `Accept: application/vnd.xet-fileinfo+json` so the
+ /// server returns XET metadata instead of redirecting; in `Http` mode the
+ /// redirect is followed and the file bytes are streamed directly.
+ pub(super) async fn resolve(
+ &self,
+ path: &str,
+ range: BytesRange,
+ mode: HfDownloadMode,
+ ) -> Result<Response<HttpBody>> {
let uri = self.uri(path);
- let url = uri.resolve_url(&self.endpoint);
-
- let req = self
- .request(http::Method::HEAD, &url, Operation::Stat)?
- .body(Buffer::new())
- .map_err(new_request_build_error)?;
+ let url = uri.resolve_url(&self.endpoint, self.repo.revision());
- let resp = self.no_redirect_client.send(req).await?;
+ let mut req = self.request(http::Method::GET, &url, Operation::Read)?;
- if resp.status().is_client_error() || resp.status().is_server_error() {
- return Err(parse_error(resp));
+ if mode == HfDownloadMode::Xet {
+ req = req.header(header::ACCEPT,
"application/vnd.xet-fileinfo+json");
}
- let hash = resp
- .headers()
- .get("X-Xet-Hash")
- .and_then(|v| v.to_str().ok())
- .filter(|s| !s.is_empty());
+ if !range.is_full() {
+ req = req.header(header::RANGE, range.to_header());
+ }
- let Some(hash) = hash else {
- return Ok(None);
- };
+ let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
+ let resp = self.info.http_client().fetch(req).await?;
- let size = resp
- .headers()
- .get("X-Linked-Size")
- .or_else(|| resp.headers().get(header::CONTENT_LENGTH))
- .and_then(|v| v.to_str().ok())
- .and_then(|s| s.parse::<u64>().ok())
- .unwrap_or(0);
+ if !resp.status().is_success() {
+ let (parts, mut body) = resp.into_parts();
+ let buf = body.to_buffer().await?;
+ return Err(parse_error(parts, buf));
+ }
- Ok(Some(XetFileInfo::new(hash.to_string(), size)))
+ Ok(resp)
}
/// Commit file changes to a git-based repo (model/dataset/space).
@@ -521,25 +587,38 @@ pub(crate) mod test_utils {
);
// Return a minimal valid JSON response for API requests
- let body = if req.uri().to_string().contains("/paths-info/")
+ let (body, content_length) = if
req.uri().to_string().contains("/paths-info/")
|| req.uri().to_string().contains("/tree/")
{
let data =
Bytes::from(r#"[{"type":"file","oid":"abc123","size":100,"path":"test.txt"}]"#);
let size = data.len() as u64;
let buffer = Buffer::from(data);
- HttpBody::new(futures::stream::iter(vec![Ok(buffer)]),
Some(size))
+ (
+ HttpBody::new(futures::stream::iter(vec![Ok(buffer)]),
Some(size)),
+ size,
+ )
} else if req.uri().to_string().contains("/commit/") {
let data = Bytes::from(r#"{}"#);
let size = data.len() as u64;
let buffer = Buffer::from(data);
- HttpBody::new(futures::stream::iter(vec![Ok(buffer)]),
Some(size))
+ (
+ HttpBody::new(futures::stream::iter(vec![Ok(buffer)]),
Some(size)),
+ size,
+ )
} else {
- HttpBody::new(futures::stream::empty(), Some(0))
+ let data = Bytes::from_static(b"hello");
+ let size = data.len() as u64;
+ let buffer = Buffer::from(data);
+ (
+ HttpBody::new(futures::stream::iter(vec![Ok(buffer)]),
Some(size)),
+ size,
+ )
};
Ok(Response::builder()
.status(StatusCode::OK)
+ .header(header::CONTENT_LENGTH, content_length)
.body(body)
.unwrap())
}
@@ -568,8 +647,8 @@ pub(crate) mod test_utils {
"/".to_string(),
None,
endpoint.to_string(),
- HttpClient::with(mock_client.clone()),
xet_session,
+ HfDownloadMode::Xet,
);
(core, mock_client)
@@ -662,12 +741,3 @@ mod tests {
Ok(())
}
}
-
-fn build_reqwest(policy: reqwest::redirect::Policy) -> Result<reqwest::Client>
{
- reqwest::Client::builder()
- .redirect(policy)
- .build()
- .map_err(|err| {
- Error::new(ErrorKind::Unexpected, "failed to build http
client").set_source(err)
- })
-}
diff --git a/core/services/hf/src/error.rs b/core/services/hf/src/error.rs
index fcad0d58b..6fcc6ccd0 100644
--- a/core/services/hf/src/error.rs
+++ b/core/services/hf/src/error.rs
@@ -17,7 +17,6 @@
use std::fmt::Debug;
-use http::Response;
use http::StatusCode;
use serde::Deserialize;
@@ -37,10 +36,8 @@ impl Debug for HfError {
}
}
-pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
- let (parts, body) = resp.into_parts();
+pub(super) fn parse_error(parts: http::response::Parts, body: Buffer) -> Error
{
let bs = body.to_bytes();
-
let message = match serde_json::from_slice::<HfError>(&bs) {
Ok(hf_error) => hf_error.error,
Err(_) => String::from_utf8_lossy(&bs).into_owned(),
@@ -78,6 +75,7 @@ pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
#[cfg(test)]
mod test {
+ use http::Response;
use http::StatusCode;
use super::*;
@@ -102,12 +100,13 @@ mod test {
let body = Buffer::from(bytes::Bytes::from(
r#"{"error":"The branch was updated since you opened this page.
Please refresh and try again."}"#,
));
- let resp = Response::builder()
+ let (parts, _) = Response::builder()
.status(StatusCode::PRECONDITION_FAILED)
- .body(body)
- .unwrap();
+ .body(())
+ .unwrap()
+ .into_parts();
- let err = parse_error(resp);
+ let err = parse_error(parts, body);
assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
assert!(err.is_temporary());
@@ -116,12 +115,13 @@ mod test {
#[test]
fn test_parse_error_other_precondition_failed_is_not_temporary() {
let body = Buffer::from(bytes::Bytes::from(r#"{"error":"etag
mismatch"}"#));
- let resp = Response::builder()
+ let (parts, _) = Response::builder()
.status(StatusCode::PRECONDITION_FAILED)
- .body(body)
- .unwrap();
+ .body(())
+ .unwrap()
+ .into_parts();
- let err = parse_error(resp);
+ let err = parse_error(parts, body);
assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
assert!(!err.is_temporary());
diff --git a/core/services/hf/src/reader.rs b/core/services/hf/src/reader.rs
index 0073d90e2..67c10b5eb 100644
--- a/core/services/hf/src/reader.rs
+++ b/core/services/hf/src/reader.rs
@@ -15,13 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use http::Response;
-use http::StatusCode;
-use http::header;
+use bytes::Buf;
use xet::xet_session::{SessionError, XetDownloadStream, XetFileInfo};
-use super::core::HfCore;
+use super::core::{HfCore, XetFileResponse};
use opendal_core::raw::*;
use opendal_core::*;
@@ -31,56 +29,20 @@ pub enum HfReader {
}
impl HfReader {
- /// Create a reader, automatically choosing between XET and HTTP.
- ///
- /// Buckets always use XET. For other repo types, a HEAD request
- /// probes for the `X-Xet-Hash` header. Files stored on XET are
- /// downloaded via the CAS protocol; all others fall back to HTTP GET.
- pub async fn try_new(core: &HfCore, path: &str, range: BytesRange) ->
Result<(Metadata, Self)> {
- if let Some(xet_file) = core.maybe_xet_file(path).await? {
- return Self::try_new_xet(core, &xet_file, range).await;
- }
-
- if core.repo.is_bucket() {
- return Err(Error::new(
- ErrorKind::Unexpected,
- "bucket file is missing XET metadata",
- ));
- }
-
- Self::try_new_http(core, path, range).await
- }
-
- pub async fn try_new_http(
- core: &HfCore,
- path: &str,
- range: BytesRange,
- ) -> Result<(Metadata, Self)> {
- let client = core.info.http_client();
- let uri = core.uri(path);
- let url = uri.resolve_url(&core.endpoint);
-
- let mut req = core.request(http::Method::GET, &url, Operation::Read)?;
-
- if !range.is_full() {
- req = req.header(header::RANGE, range.to_header());
- }
-
- let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
-
- let resp = client.fetch(req).await?;
- let status = resp.status();
-
- match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((
- parse_into_metadata(path, resp.headers())?,
- Self::Http(resp.into_body()),
- )),
- _ => {
- let (part, mut body) = resp.into_parts();
- let buf = body.to_buffer().await?;
- Err(super::error::parse_error(Response::from_parts(part, buf)))
- }
+ pub async fn try_new(core: &HfCore, path: &str, range: BytesRange) ->
Result<(RpRead, Self)> {
+ let resp = core.resolve(path, range, core.download_mode).await?;
+ if resp.headers().contains_key("x-xet-hash") {
+ let (_, mut body) = resp.into_parts();
+ let buf = body.to_buffer().await?;
+ let info: XetFileResponse =
+
serde_json::from_reader(buf.reader()).map_err(new_json_deserialize_error)?;
+ let metadata =
Metadata::new(EntryMode::FILE).with_content_length(info.size);
+ let reader =
+ Self::try_new_xet(core, &XetFileInfo::new(info.hash,
info.size), range).await?;
+ Ok((RpRead::new(metadata), reader))
+ } else {
+ let metadata = parse_into_metadata(path, resp.headers())?;
+ Ok((RpRead::new(metadata), Self::Http(resp.into_body())))
}
}
@@ -88,7 +50,7 @@ impl HfReader {
core: &HfCore,
file_info: &XetFileInfo,
range: BytesRange,
- ) -> Result<(Metadata, Self)> {
+ ) -> Result<Self> {
let group = core.xet_download_group().await?;
let xet_range = if range.is_full() {
@@ -110,13 +72,7 @@ impl HfReader {
.set_source(err)
})?;
stream.start();
-
- let total_size = file_info.file_size.unwrap_or_default();
- let metadata = Metadata::new(EntryMode::FILE)
- .with_content_length(total_size)
- .with_etag(file_info.hash().to_string());
-
- Ok((metadata, Self::Xet(stream)))
+ Ok(Self::Xet(stream))
}
}
@@ -139,7 +95,13 @@ impl oio::Read for HfReader {
#[cfg(test)]
mod tests {
- use super::super::backend::test_utils::{gpt2_operator, mbpp_operator};
+ use super::super::backend::test_utils::{gpt2_operator, mbpp_operator,
testing_dataset_core};
+ use super::super::core::test_utils::create_test_core;
+ use super::super::core::{CommitFile, DeletedFile};
+ use super::super::uri::HfRepoType;
+ use super::*;
+ use bytes::Bytes;
+ use opendal_core::raw::oio::Read;
/// Parquet magic bytes: "PAR1"
const PARQUET_MAGIC: &[u8] = b"PAR1";
@@ -152,6 +114,28 @@ mod tests {
.expect("config.json should be valid JSON");
}
+ #[tokio::test]
+ async fn test_http_read_returns_metadata() -> Result<()> {
+ let (core, _) = create_test_core(
+ HfRepoType::Model,
+ "test-user/test-repo",
+ "main",
+ "https://huggingface.co";,
+ );
+
+ let (rp, mut reader) = HfReader::try_new(&core, "test.txt",
BytesRange::default()).await?;
+ let metadata = rp.metadata().expect("read metadata must be returned");
+
+ assert_eq!(metadata.mode(), EntryMode::FILE);
+ assert_eq!(metadata.content_length(), 5);
+ assert!(matches!(reader, HfReader::Http(_)));
+
+ let chunk = reader.read().await?;
+ assert_eq!(chunk.to_bytes(), Bytes::from_static(b"hello"));
+
+ Ok(())
+ }
+
/// Exercises the XET download code path against a public dataset known to
/// have XET-stored files. Behavior tests cannot reliably cover this path
/// because the test dataset may not contain any XET files.
@@ -169,6 +153,63 @@ mod tests {
assert_eq!(&bytes[bytes.len() - 4..], PARQUET_MAGIC);
}
+ /// Verifies that a non-XET file (plain git blob) read in Xet mode falls
back
+ /// to the HTTP body path rather than erroring. Uploads a small file via
the
+ /// git commit API (which does not go through XET), then reads it back.
+ /// Requires HF_OPENDAL_DATASET and HF_OPENDAL_TOKEN.
+ #[tokio::test]
+ #[ignore = "requires network access"]
+ async fn test_xet_mode_falls_back_to_http_for_non_xet_file() {
+ use base64::Engine;
+
+ let core = testing_dataset_core();
+ let content = b"non-xet fallback test content";
+ let path = "tests/non-xet-fallback.txt";
+
+ core.commit_git(
+ vec![CommitFile {
+ path: path.to_string(),
+ content: base64::prelude::BASE64_STANDARD.encode(content),
+ encoding: "base64".to_string(),
+ }],
+ vec![],
+ vec![],
+ vec![],
+ )
+ .await
+ .expect("commit should succeed");
+
+ let (_, mut reader) = HfReader::try_new(&core, path,
BytesRange::default())
+ .await
+ .expect("reading non-XET file in Xet mode should succeed via HTTP
fallback");
+
+ assert!(
+ matches!(reader, HfReader::Http(_)),
+ "expected HTTP reader for non-XET file"
+ );
+
+ let mut buf = Vec::new();
+ loop {
+ let chunk: Buffer = reader.read().await.expect("read chunk should
succeed");
+ if chunk.is_empty() {
+ break;
+ }
+ buf.extend_from_slice(&chunk.to_bytes());
+ }
+ assert_eq!(buf, content);
+
+ core.commit_git(
+ vec![],
+ vec![],
+ vec![DeletedFile {
+ path: path.to_string(),
+ }],
+ vec![],
+ )
+ .await
+ .ok();
+ }
+
/// Exercises XET range reads (XetDownloadStream with a byte range).
#[tokio::test]
#[ignore = "requires network access"]
diff --git a/core/services/hf/src/uri.rs b/core/services/hf/src/uri.rs
index 452d2314e..32bdb08b9 100644
--- a/core/services/hf/src/uri.rs
+++ b/core/services/hf/src/uri.rs
@@ -16,58 +16,11 @@
// under the License.
use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
-use serde::Deserialize;
-use serde::Serialize;
use super::HUGGINGFACE_SCHEME;
+pub use super::core::HfRepoType;
use opendal_core::raw::*;
-/// Repository type of Huggingface. Supports `model`, `dataset`, `space`, and
`bucket`.
-/// [Reference](https://huggingface.co/docs/hub/repositories)
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
-#[serde(rename_all = "lowercase")]
-pub enum HfRepoType {
- #[default]
- Model,
- Dataset,
- Space,
- Bucket,
-}
-
-impl HfRepoType {
- pub fn parse(s: &str) -> opendal_core::Result<Self> {
- match s.to_lowercase().replace(' ', "").as_str() {
- "model" | "models" => Ok(Self::Model),
- "dataset" | "datasets" => Ok(Self::Dataset),
- "space" | "spaces" => Ok(Self::Space),
- "bucket" | "buckets" => Ok(Self::Bucket),
- other => Err(opendal_core::Error::new(
- opendal_core::ErrorKind::ConfigInvalid,
- format!("unknown repo type: {other}"),
- )
- .with_context("service", HUGGINGFACE_SCHEME)),
- }
- }
-
- pub fn as_str(&self) -> &'static str {
- match self {
- Self::Model => "model",
- Self::Dataset => "dataset",
- Self::Space => "space",
- Self::Bucket => "bucket",
- }
- }
-
- pub fn as_plural_str(&self) -> &'static str {
- match self {
- Self::Model => "models",
- Self::Dataset => "datasets",
- Self::Space => "spaces",
- Self::Bucket => "buckets",
- }
- }
-}
-
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HfRepo {
pub repo_type: HfRepoType,
@@ -277,9 +230,12 @@ impl HfUri {
self.repo.revision()
}
- /// Build the resolve URL for this URI.
- pub fn resolve_url(&self, endpoint: &str) -> String {
- let revision = percent_encode_revision(self.revision());
+ /// Build the resolve URL for this URI using an explicit revision (e.g. a
commit OID).
+ ///
+ /// Pinning to a specific commit OID avoids CDN consistency lag that can
occur
+ /// when using a branch name like "main" immediately after a commit.
+ pub fn resolve_url(&self, endpoint: &str, revision: &str) -> String {
+ let revision = percent_encode_revision(revision);
let path = percent_encode_path(&self.path);
match self.repo.repo_type {
HfRepoType::Model => {
@@ -559,7 +515,7 @@ mod tests {
#[test]
fn test_bucket_resolve_url() {
let p = resolve("buckets/user/bucket/file.txt");
- let url = p.resolve_url("https://huggingface.co";);
+ let url = p.resolve_url("https://huggingface.co";, p.revision());
assert_eq!(
url,
"https://huggingface.co/buckets/user/bucket/resolve/file.txt";
