refactor: replace custom XET sink with OpenDAL ObjectStore for HF URLs

Replace the custom HfBucketSinkNode (1,050 lines of XET-specific code)
with a standard ObjectStore implementation backed by OpenDAL's HF service.

HF URLs now flow through the same FileSink path as S3/GCS/Azure,
requiring only a thin build_hf() builder in a new hf.rs module.

Key changes:
- Add crates/polars-io/src/cloud/hf.rs: HF URL parsing, token
  extraction, and OpenDAL ObjectStore construction (~175 lines)
- Wire CloudType::Hf in object_store_setup.rs to call build_hf(),
  matching the pattern used by build_aws/build_gcp/build_azure
- Delete custom sink: hf_bucket/ directory (4 files, 721 lines),
  HfBucketSinkNode (260 lines), IR lowering special-case,
  PhysNodeKind::HfBucketSink variant
- Rename feature flag hf_bucket_sink -> hf across 9 Cargo.toml files
- Bump object_store_opendal compatibility from object_store 0.12 to 0.13

Dependencies: opendal + object_store_opendal (local path deps for now,
will switch to published crate versions once apache/opendal#7185 ships).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: davanstrien

Cargo.lock

@@ -73,6 +73,8 @@ num-derive = "0.4.2"
 num-traits = "0.2"
 numpy = "0.28"
 object_store = { version = "0.13.1", default-features = false, features = ["fs"] }
+object_store_opendal = { version = "0.55.0", default-features = false }
+opendal = { version = "0.55.0", default-features = false }
 parking_lot = "0.12"
 percent-encoding = "2.3"
 pin-project-lite = "0.2"
@@ -164,6 +166,8 @@ collapsible_if = "allow"
 # simd-json = { git = "https://github.com/ritchie46/simd-json", branch = "alignment" }
 tikv-jemallocator = { git = "https://github.com/pola-rs/jemallocator", rev = "c7991e5bb6b3e9f79db6b0f48dcda67c5c3d2936" }
 object_store = { git = "https://github.com/kdn36/arrow-rs-object-store", branch = "feat_checksum_crc64" }
+opendal = { path = "opendal/core" }
+object_store_opendal = { path = "opendal/integrations/object_store" }
 color-backtrace = { git = "https://github.com/orlp/color-backtrace", rev = "bb62ccf1e9eb1f6b7af5f16acff1fd7151a876dd" }
 
 [profile.mindebug-dev]

Cargo.toml

@@ -54,8 +54,8 @@ tokio = { workspace = true, features = ["fs", "net", "rt-multi-thread", "time",
 zmij = { workspace = true, optional = true }
 zstd = { workspace = true, optional = true }
 
-hf-xet = { git = "https://github.com/huggingface/xet-core", rev = "cacd713", optional = true }
-xet-client = { git = "https://github.com/huggingface/xet-core", rev = "cacd713", optional = true }
+opendal = { workspace = true, features = ["services-hf"], optional = true }
+object_store_opendal = { workspace = true, optional = true }
 
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
 fs4 = { version = "0.13", features = ["sync"], optional = true }
@@ -150,7 +150,7 @@ http = ["object_store/http", "cloud"]
 temporal = ["dtype-datetime", "dtype-date", "dtype-time"]
 simd = []
 python = ["pyo3", "polars-error/python", "polars-utils/python"]
-hf_bucket_sink = ["cloud", "parquet", "dep:hf-xet", "dep:xet-client"]
+hf = ["cloud", "dep:opendal", "dep:object_store_opendal"]
 allow_unused = []
 
 [package.metadata.docs.rs]

crates/polars-io/Cargo.toml

@@ -0,0 +1,175 @@
+//! Hugging Face cloud storage support via OpenDAL.
+//!
+//! Provides an [`ObjectStore`] implementation for `hf://` URLs by bridging
+//! OpenDAL's HF backend through `object_store_opendal`.
+//!
+//! Gated behind `#[cfg(feature = "hf")]`.
+
+use std::sync::Arc;
+
+use object_store::ObjectStore;
+use polars_error::{PolarsResult, polars_bail, polars_err, to_compute_err};
+use polars_utils::pl_path::PlRefPath;
+
+use super::options::CloudOptions;
+
+/// Parse an `hf://` URL and build an [`ObjectStore`] backed by OpenDAL.
+///
+/// Supported URL formats:
+/// - `hf://buckets/<namespace>/<name>[/<path>]`
+/// - `hf://datasets/<namespace>/<name>[/<path>]`
+/// - `hf://models/<namespace>/<name>[/<path>]`
+pub fn build_hf(
+    url: PlRefPath,
+    options: Option<&CloudOptions>,
+) -> PolarsResult<Arc<dyn ObjectStore>> {
+    let after_scheme = url.strip_scheme();
+    let (repo_type_plural, rest) = after_scheme
+        .split_once('/')
+        .ok_or_else(|| polars_err!(ComputeError: "invalid hf:// URL: {}", url.as_str()))?;
+
+    // hf:// URLs use plural form ("buckets", "datasets", "models")
+    // but OpenDAL expects singular ("bucket", "dataset", "model")
+    let repo_type: &str = repo_type_plural
+        .strip_suffix('s')
+        .unwrap_or(repo_type_plural);
+
+    // Extract repo_id (namespace/name) from the remaining path
+    let parts = rest.splitn(3, '/').collect::<Vec<&str>>();
+    if parts.len() < 2 || parts[0].is_empty() || parts[1].is_empty() {
+        polars_bail!(
+            ComputeError:
+            "invalid hf:// URL: expected hf://<type>/<namespace>/<name>[/path], got: {}",
+            url.as_str()
+        );
+    }
+    let repo_id = format!("{}/{}", parts[0], parts[1]);
+
+    let token = extract_hf_token(options)?;
+
+    let builder = opendal::services::Hf::default()
+        .repo_type(repo_type)
+        .repo_id(&repo_id)
+        .token(&token);
+
+    let op = opendal::Operator::new(builder)
+        .map_err(to_compute_err)?
+        .finish();
+
+    Ok(Arc::new(object_store_opendal::OpendalStore::new(op)) as Arc<dyn ObjectStore>)
+}
+
+/// Extract an HF token from cloud options, environment, or cached file.
+///
+/// Resolution order:
+/// 1. `storage_options` / CloudOptions HTTP Authorization header
+/// 2. `HF_TOKEN` environment variable
+/// 3. Cached token at `$HF_HOME/token` (default: `~/.cache/huggingface/token`)
+fn extract_hf_token(cloud_options: Option<&CloudOptions>) -> PolarsResult<String> {
+    #[cfg(feature = "http")]
+    if let Some(opts) = cloud_options {
+        if let Some(super::options::CloudConfig::Http { headers }) = &opts.config {
+            for (key, value) in headers {
+                if key.eq_ignore_ascii_case("authorization") {
+                    if let Some(token) = value.strip_prefix("Bearer ") {
+                        return Ok(token.to_string());
+                    }
+                }
+            }
+        }
+    }
+
+    #[cfg(not(feature = "http"))]
+    let _ = cloud_options;
+
+    if let Ok(token) = std::env::var("HF_TOKEN") {
+        if !token.is_empty() {
+            return Ok(token);
+        }
+    }
+
+    let hf_home = std::env::var("HF_HOME");
+    let hf_home = hf_home.as_deref().unwrap_or("~/.cache/huggingface");
+    let hf_home = crate::path_utils::resolve_homedir(hf_home);
+    let cached_token_path = hf_home.join("token");
+
+    if let Ok(bytes) = std::fs::read(&cached_token_path) {
+        if let Ok(token) = String::from_utf8(bytes) {
+            let token = token.trim().to_string();
+            if !token.is_empty() {
+                return Ok(token);
+            }
+        }
+    }
+
+    polars_bail!(
+        ComputeError:
+        "no HF token found: set HF_TOKEN env var, pass via storage_options, \
+         or login with `huggingface-cli login`"
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_token_from_env() {
+        let original = std::env::var("HF_TOKEN").ok();
+        std::env::set_var("HF_TOKEN", "hf_test_token_123");
+
+        let result = extract_hf_token(None);
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap(), "hf_test_token_123");
+
+        match original {
+            Some(v) => std::env::set_var("HF_TOKEN", v),
+            None => std::env::remove_var("HF_TOKEN"),
+        }
+    }
+
+    #[test]
+    fn test_empty_token_skipped() {
+        let original = std::env::var("HF_TOKEN").ok();
+        std::env::set_var("HF_TOKEN", "");
+
+        let result = extract_hf_token(None);
+        if let Ok(token) = &result {
+            assert!(!token.is_empty());
+        }
+
+        match original {
+            Some(v) => std::env::set_var("HF_TOKEN", v),
+            None => std::env::remove_var("HF_TOKEN"),
+        }
+    }
+
+    #[test]
+    fn test_build_hf_valid_bucket_url() {
+        std::env::set_var("HF_TOKEN", "hf_test");
+        let url = PlRefPath::new("hf://buckets/myorg/mybucket/path/file.parquet");
+        let result = build_hf(url, None);
+        // Builder succeeds (actual I/O would fail without a real token,
+        // but the ObjectStore is constructed)
+        assert!(result.is_ok());
+        std::env::remove_var("HF_TOKEN");
+    }
+
+    #[test]
+    fn test_build_hf_valid_dataset_url() {
+        std::env::set_var("HF_TOKEN", "hf_test");
+        let url = PlRefPath::new("hf://datasets/user/dataset-name/train.parquet");
+        let result = build_hf(url, None);
+        assert!(result.is_ok());
+        std::env::remove_var("HF_TOKEN");
+    }
+
+    #[test]
+    fn test_build_hf_invalid_url_no_repo() {
+        std::env::set_var("HF_TOKEN", "hf_test");
+        let url = PlRefPath::new("hf://buckets/only-namespace");
+        let result = build_hf(url, None);
+        assert!(result.is_err());
+        std::env::remove_var("HF_TOKEN");
+    }
+}