From 2b4f03de55181aafdf08c511441394ccd3455e71 Mon Sep 17 00:00:00 2001 From: comphead Date: Thu, 22 Oct 2026 17:33:38 -0700 Subject: [PATCH 1/2] feat(storage): Add HDFS support via opendal services-hdfs (libhdfs) Add an `opendal-hdfs` feature that serves `hdfs://` paths through OpenDAL's `services-hdfs`, libhdfs over JNI, next to `hdfs-native` as #1130 agreed. Only the transport differs. Hadoop's Java client reads its configuration from CLASSPATH and resolves nameservices, HA, Kerberos and `fs.defaultFS` itself, so the backend maps a path to the NameNode URI `FileSystem.get` takes and caches one operator per NameNode. The `hdfs-native` backend and the `iceberg` crate are unchanged. --- .github/workflows/ci.yml | 13 ++ Cargo.lock | 110 +++++++++++++++ crates/storage/opendal/Cargo.toml | 1 + crates/storage/opendal/README.md | 3 + crates/storage/opendal/public-api.txt | 11 ++ crates/storage/opendal/src/hdfs.rs | 132 ++++++++++++++++++ crates/storage/opendal/src/lib.rs | 34 +++++ .../opendal/tests/file_io_libhdfs_test.rs | 117 ++++++++++++++++ dev/hdfs/client/core-site.xml | 32 +++++ 9 files changed, 453 insertions(+) create mode 100644 crates/storage/opendal/src/hdfs.rs create mode 100644 crates/storage/opendal/tests/file_io_libhdfs_test.rs create mode 100644 dev/hdfs/client/core-site.xml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cbff03bb1f..0c4d2f4d67 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -253,6 +253,19 @@ jobs: if: matrix.test-suite.name == 'default' run: make docker-up + # `opendal-hdfs` links libjvm into every test binary. Its tests run + # Hadoop's Java client from the HDFS fixture's jars. + - name: Set up the JVM for libhdfs + shell: bash + run: | + echo "LD_LIBRARY_PATH=$JAVA_HOME/lib/server${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}" >> "$GITHUB_ENV" + if [ "${{ matrix.test-suite.name }}" = "default" ]; then + docker compose -f dev/docker-compose.yaml cp hdfs-namenode:/opt/hadoop/share/hadoop "$RUNNER_TEMP/hadoop" + # libhdfs does not expand wildcards, so list each jar. + jars=$(find "$RUNNER_TEMP/hadoop/common" "$RUNNER_TEMP/hadoop/hdfs" -name '*.jar' ! -name '*-tests.jar' ! -name '*-sources.jar' | sort | paste -sd: -) + echo "CLASSPATH=$PWD/dev/hdfs/client:$jars" >> "$GITHUB_ENV" + fi + - name: Run tests shell: bash env: diff --git a/Cargo.lock b/Cargo.lock index a2d7370a0d..0f864e7efe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -497,6 +497,18 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-channel" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" +dependencies = [ + "concurrent-queue", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + [[package]] name = "async-lock" version = "3.4.2" @@ -519,6 +531,12 @@ dependencies = [ "syn", ] +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + [[package]] name = "async-trait" version = "0.1.89" @@ -1157,6 +1175,19 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "blocking" +version = "1.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a70e4329df6cb94385eed412ec92375c3cdd8a6e502493d1229b6414e4036dfa" +dependencies = [ + "async-channel", + "async-task", + "futures-io", + "futures-lite", + "piper", +] + [[package]] name = "bnum" version = "0.12.1" @@ -1468,6 +1499,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "concurrent-queue" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "const-oid" version = "0.9.6" @@ -2289,6 +2329,16 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" +[[package]] +name = "futures-lite" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" +dependencies = [ + "futures-core", + "pin-project-lite", +] + [[package]] name = "futures-macro" version = "0.3.32" @@ -2580,6 +2630,30 @@ dependencies = [ "whoami 2.1.2", ] +[[package]] +name = "hdfs-sys" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18138c3dacdf0e95f834369f996b21d2b6fa99e71787ef45e2ece09b37bd1dd1" +dependencies = [ + "cc", + "java-locator", +] + +[[package]] +name = "hdrs" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1ac22889be3d758d8168395e2ab70374afb4b1959caa9586ab42e6dd9f454d" +dependencies = [ + "blocking", + "errno", + "futures", + "hdfs-sys", + "libc", + "log", +] + [[package]] name = "heapify" version = "0.2.0" @@ -3322,6 +3396,15 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "java-locator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09c46c1fe465c59b1474e665e85e1256c3893dd00927b8d55f63b09044c1e64f" +dependencies = [ + "glob", +] + [[package]] name = "jiff" version = "0.2.31" @@ -4105,6 +4188,7 @@ dependencies = [ "opendal-service-azdls", "opendal-service-fs", "opendal-service-gcs", + "opendal-service-hdfs", "opendal-service-hdfs-native", "opendal-service-hf", "opendal-service-oss", @@ -4260,6 +4344,21 @@ dependencies = [ "tokio", ] +[[package]] +name = "opendal-service-hdfs" +version = "0.58.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00570d1cc0c3f00db60bc8aaa63aa304e176285264e8674ca11897afaf509629" +dependencies = [ + "bytes", + "futures", + "hdrs", + "log", + "opendal-core", + "serde", + "tokio", +] + [[package]] name = "opendal-service-hdfs-native" version = "0.58.1" @@ -4585,6 +4684,17 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "piper" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" +dependencies = [ + "atomic-waker", + "fastrand", + "futures-io", +] + [[package]] name = "pkcs1" version = "0.7.5" diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index 83541e89c5..dacd02ccb3 100644 --- a/crates/storage/opendal/Cargo.toml +++ b/crates/storage/opendal/Cargo.toml @@ -42,6 +42,7 @@ opendal-all = [ opendal-azdls = ["opendal/services-azdls"] opendal-fs = ["opendal/services-fs"] opendal-gcs = ["opendal/services-gcs"] +opendal-hdfs = ["opendal/services-hdfs"] opendal-hdfs-native = [ "opendal/services-hdfs-native", "dep:tokio", diff --git a/crates/storage/opendal/README.md b/crates/storage/opendal/README.md index ac3a8b62ef..3211ed2f3d 100644 --- a/crates/storage/opendal/README.md +++ b/crates/storage/opendal/README.md @@ -33,6 +33,7 @@ OpenDAL-based storage backend implementations for [Apache Iceberg Rust](https:// | Alibaba Cloud OSS | `opendal-oss` | 🧪 Experimental | Alibaba Cloud Object Storage Service | | Azure Datalake | `opendal-azdls` | 🧪 Experimental | Azure Datalake Storage v2 | | HDFS | `opendal-hdfs-native` | 🧪 Experimental | HDFS via the native Rust client (`hdfs-native`) | +| HDFS (libhdfs) | `opendal-hdfs` | 🧪 Experimental | HDFS via libhdfs and Hadoop's Java client (JNI) | You can enable all stable storage backends at once using the `opendal-all` feature flag. @@ -40,6 +41,8 @@ You can enable all stable storage backends at once using the `opendal-all` featu > `opendal-hdfs-native` is experimental and not included in `opendal-all`. A portless path authority such as `hdfs://nameservice1/warehouse` must be declared in properties, e.g. `hdfs.name-node.nameservice1 = nn1:8020,nn2:8020`: unlike Hadoop, nameservices defined only in `$HADOOP_CONF_DIR` are not resolved, and a portless host is not dialed on the default port 8020. +> `opendal-hdfs` is experimental and not included in `opendal-all`. Use it through `OpenDalStorageFactory::Hdfs`, since `OpenDalResolvingStorageFactory` does not route to it. Hadoop's Java client reads its configuration from `CLASSPATH`, so nameservices, HA and Kerberos work as in Hadoop, and no `hdfs.*` or `hadoop.*` property applies. Building needs a JDK, and binaries need `$JAVA_HOME/lib/server` on the library path. Unless the process already runs a JVM, `CLASSPATH` must list the Hadoop configuration directory and each Hadoop jar, e.g. from `hadoop classpath --glob`. + ## Usage Add the crate to your `Cargo.toml` with the feature flags for the backends you need: diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index 797d90d1e8..33b6a57094 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -8,6 +8,7 @@ pub iceberg_storage_opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc +pub iceberg_storage_opendal::OpenDalStorage::Hdfs(iceberg_storage_opendal::HdfsStorage) pub iceberg_storage_opendal::OpenDalStorage::HdfsNative(iceberg_storage_opendal::HdfsNativeStorage) pub iceberg_storage_opendal::OpenDalStorage::Hf pub iceberg_storage_opendal::OpenDalStorage::Hf::client_config: iceberg_storage_opendal::OpenDalClientConfig @@ -48,6 +49,7 @@ pub enum iceberg_storage_opendal::OpenDalStorageFactory pub iceberg_storage_opendal::OpenDalStorageFactory::Azdls pub iceberg_storage_opendal::OpenDalStorageFactory::Fs pub iceberg_storage_opendal::OpenDalStorageFactory::Gcs +pub iceberg_storage_opendal::OpenDalStorageFactory::Hdfs pub iceberg_storage_opendal::OpenDalStorageFactory::HdfsNative pub iceberg_storage_opendal::OpenDalStorageFactory::Hf pub iceberg_storage_opendal::OpenDalStorageFactory::Memory @@ -80,6 +82,15 @@ impl serde_core::ser::Serialize for iceberg_storage_opendal::HdfsNativeStorage pub fn iceberg_storage_opendal::HdfsNativeStorage::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_opendal::HdfsNativeStorage pub fn iceberg_storage_opendal::HdfsNativeStorage::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_storage_opendal::HdfsStorage +impl core::clone::Clone for iceberg_storage_opendal::HdfsStorage +pub fn iceberg_storage_opendal::HdfsStorage::clone(&self) -> iceberg_storage_opendal::HdfsStorage +impl core::fmt::Debug for iceberg_storage_opendal::HdfsStorage +pub fn iceberg_storage_opendal::HdfsStorage::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg_storage_opendal::HdfsStorage +pub fn iceberg_storage_opendal::HdfsStorage::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_opendal::HdfsStorage +pub fn iceberg_storage_opendal::HdfsStorage::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_storage_opendal::OpenDalClientConfig impl iceberg_storage_opendal::OpenDalClientConfig pub fn iceberg_storage_opendal::OpenDalClientConfig::from_properties(properties: &std::collections::hash::map::HashMap) -> iceberg::error::Result diff --git a/crates/storage/opendal/src/hdfs.rs b/crates/storage/opendal/src/hdfs.rs new file mode 100644 index 0000000000..fa9c8d4a63 --- /dev/null +++ b/crates/storage/opendal/src/hdfs.rs @@ -0,0 +1,132 @@ +// 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. + +//! HDFS storage backend via OpenDAL's `services-hdfs` (libhdfs over JNI). +//! Hadoop's Java client reads its own configuration from `CLASSPATH`. + +use std::collections::HashMap; +use std::sync::{Arc, PoisonError, RwLock}; + +use iceberg::{Error, ErrorKind, Result}; +use opendal::Operator; +use opendal::services::HdfsConfig; +use serde::{Deserialize, Serialize}; + +use crate::OpenDalClientConfig; +use crate::utils::from_opendal_error; + +/// State of [`OpenDalStorage::Hdfs`](crate::OpenDalStorage::Hdfs): an operator +/// per NameNode, shared by clones, because building one connects. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct HdfsStorage { + #[serde(skip)] + pub(crate) operators: Arc>>, + #[serde(default)] + pub(crate) client_config: OpenDalClientConfig, +} + +impl HdfsStorage { + pub(crate) fn new(client_config: OpenDalClientConfig) -> Self { + Self { + operators: Arc::default(), + client_config, + } + } +} + +/// The NameNode URI for a path, as Hadoop's `FileSystem.get` takes it, and +/// the path relative to it. `hdfs:///` resolves to `fs.defaultFS`. +pub(crate) fn hdfs_name_node(path: &str) -> Result<(String, &str)> { + let rest = path.strip_prefix("hdfs://").ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}, expected scheme `hdfs://`"), + ) + })?; + let (authority, relative_path) = rest.split_once('/').unwrap_or((rest, "")); + // OpenDAL paths must not start with `/`. + Ok(( + format!("hdfs://{authority}/"), + relative_path.trim_start_matches('/'), + )) +} + +/// Creates an operator for the path, reusing the cached one for its NameNode. +pub(crate) fn hdfs_create_operator<'a>( + path: &'a str, + operators: &RwLock>, +) -> Result<(Operator, &'a str)> { + let (name_node, relative_path) = hdfs_name_node(path)?; + if let Some(op) = operators + .read() + .unwrap_or_else(PoisonError::into_inner) + .get(&name_node) + { + return Ok((op.clone(), relative_path)); + } + + // Built outside the lock, since connecting may start a JVM. A racing + // caller keeps the operator inserted first. + let mut config = HdfsConfig::default(); + config.name_node = Some(name_node.clone()); + let op = Operator::from_config(config).map_err(from_opendal_error)?; + let op = operators + .write() + .unwrap_or_else(PoisonError::into_inner) + .entry(name_node) + .or_insert(op) + .clone(); + Ok((op, relative_path)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::OpenDalStorage; + + #[test] + fn test_hdfs_name_node() { + for (path, name_node, relative_path) in [ + ("hdfs://nn:8020/a/b", "hdfs://nn:8020/", "a/b"), + ("hdfs://ns1//a", "hdfs://ns1/", "a"), + ("hdfs:///a/b", "hdfs:///", "a/b"), + ("hdfs://nn:8020", "hdfs://nn:8020/", ""), + ] { + assert_eq!( + hdfs_name_node(path).unwrap(), + (name_node.to_string(), relative_path), + "{path}" + ); + } + assert!(hdfs_name_node("s3://bucket/a").is_err()); + } + + #[test] + fn test_hdfs_delete_stream_keys() { + let storage = OpenDalStorage::Hdfs(HdfsStorage::new(OpenDalClientConfig::default())); + + // NameNodes that differ only by port must not share a deleter. + assert_ne!( + storage.batch_key_for_path("hdfs://nn:8020/a"), + storage.batch_key_for_path("hdfs://nn:9000/a") + ); + assert_eq!( + storage.relativize_path("hdfs://nn:8020/a/b").unwrap(), + "a/b" + ); + } +} diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 693f4c18b3..4347f9a5ae 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -77,6 +77,14 @@ cfg_if! { } } +cfg_if! { + if #[cfg(feature = "opendal-hdfs")] { + mod hdfs; + use hdfs::*; + pub use hdfs::HdfsStorage; + } +} + cfg_if! { if #[cfg(feature = "opendal-hdfs-native")] { mod hdfs_native; @@ -194,6 +202,9 @@ pub enum OpenDalStorageFactory { /// GCS storage factory. #[cfg(feature = "opendal-gcs")] Gcs, + /// HDFS storage factory through libhdfs. + #[cfg(feature = "opendal-hdfs")] + Hdfs, /// HDFS storage factory. #[cfg(feature = "opendal-hdfs-native")] HdfsNative, @@ -247,6 +258,10 @@ impl StorageFactory for OpenDalStorageFactory { config: gcs_config_parse(config.props().clone())?.into(), client_config, })), + #[cfg(feature = "opendal-hdfs")] + OpenDalStorageFactory::Hdfs => Ok(Arc::new(OpenDalStorage::Hdfs(HdfsStorage::new( + client_config, + )))), #[cfg(feature = "opendal-hdfs-native")] OpenDalStorageFactory::HdfsNative => Ok(Arc::new(OpenDalStorage::HdfsNative( HdfsNativeStorage::new( @@ -277,6 +292,7 @@ impl StorageFactory for OpenDalStorageFactory { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs"), not(feature = "opendal-hdfs-native"), ))] _ => Err(Error::new( @@ -339,6 +355,9 @@ pub enum OpenDalStorage { #[serde(default)] client_config: OpenDalClientConfig, }, + /// HDFS storage variant through libhdfs; see [`HdfsStorage`]. + #[cfg(feature = "opendal-hdfs")] + Hdfs(HdfsStorage), /// HDFS storage variant; see [`HdfsNativeStorage`]. #[cfg(feature = "opendal-hdfs-native")] HdfsNative(HdfsNativeStorage), @@ -457,6 +476,8 @@ impl OpenDalStorage { )); } } + #[cfg(feature = "opendal-hdfs")] + OpenDalStorage::Hdfs(storage) => hdfs_create_operator(path, &storage.operators)?, #[cfg(feature = "opendal-hdfs-native")] OpenDalStorage::HdfsNative(storage) => { hdfs_native_create_operator(path, &storage.config, &storage.operators)? @@ -485,6 +506,7 @@ impl OpenDalStorage { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs"), not(feature = "opendal-hdfs-native"), ))] _ => { @@ -527,6 +549,8 @@ impl OpenDalStorage { OpenDalStorage::Azdls { client_config, .. } => client_config, #[cfg(feature = "opendal-hf")] OpenDalStorage::Hf { client_config, .. } => client_config, + #[cfg(feature = "opendal-hdfs")] + OpenDalStorage::Hdfs(storage) => &storage.client_config, #[cfg(feature = "opendal-hdfs-native")] OpenDalStorage::HdfsNative(storage) => &storage.client_config, #[cfg(all( @@ -537,6 +561,7 @@ impl OpenDalStorage { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs"), not(feature = "opendal-hdfs-native"), ))] _ => unreachable!(), @@ -553,6 +578,10 @@ impl OpenDalStorage { OpenDalStorage::Hf { .. } => hf_batch_key(path), // The URL host alone would merge distinct NameNodes that differ // only by port; key by the effective NameNode instead. + #[cfg(feature = "opendal-hdfs")] + OpenDalStorage::Hdfs(_) => hdfs_name_node(path) + .map(|(name_node, _)| name_node) + .unwrap_or_default(), #[cfg(feature = "opendal-hdfs-native")] OpenDalStorage::HdfsNative(storage) => hdfs_native_batch_key(&storage.config, path), _ => url::Url::parse(path) @@ -614,6 +643,8 @@ impl OpenDalStorage { )) } } + #[cfg(feature = "opendal-hdfs")] + OpenDalStorage::Hdfs(_) => Ok(hdfs_name_node(path)?.1), #[cfg(feature = "opendal-hdfs-native")] OpenDalStorage::HdfsNative(storage) => { let (_, relative_path) = hdfs_native_effective_name_node(&storage.config, path)?; @@ -658,6 +689,7 @@ impl OpenDalStorage { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs"), not(feature = "opendal-hdfs-native"), ))] _ => Err(Error::new( @@ -969,6 +1001,8 @@ mod tests { OpenDalStorageFactory::Azdls, #[cfg(feature = "opendal-hf")] OpenDalStorageFactory::Hf, + #[cfg(feature = "opendal-hdfs")] + OpenDalStorageFactory::Hdfs, #[cfg(feature = "opendal-hdfs-native")] OpenDalStorageFactory::HdfsNative, ]; diff --git a/crates/storage/opendal/tests/file_io_libhdfs_test.rs b/crates/storage/opendal/tests/file_io_libhdfs_test.rs new file mode 100644 index 0000000000..0a102a5af2 --- /dev/null +++ b/crates/storage/opendal/tests/file_io_libhdfs_test.rs @@ -0,0 +1,117 @@ +// 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. + +//! Integration tests for HDFS FileIO via OpenDAL `services-hdfs` (libhdfs). +//! +//! These tests need the HDFS fixture in `dev/docker-compose.yaml` and +//! Hadoop's Java client: `CLASSPATH` lists `dev/hdfs/client` and then each +//! Hadoop jar, and `$JAVA_HOME/lib/server` is on the library path. + +#[cfg(feature = "opendal-hdfs")] +mod tests { + use std::sync::Arc; + + use bytes::Bytes; + use futures::StreamExt; + use iceberg::io::{FileIO, FileIOBuilder}; + use iceberg_storage_opendal::OpenDalStorageFactory; + use iceberg_test_utils::{get_hdfs_endpoint, normalize_test_name_with_parts, set_up}; + + fn get_file_io() -> FileIO { + set_up(); + FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Hdfs)).build() + } + + fn test_path(suffix: &str) -> String { + format!( + "{}/{}", + get_hdfs_endpoint(), + normalize_test_name_with_parts!(suffix) + ) + } + + #[tokio::test] + async fn test_file_io_libhdfs_write_and_read() { + let file_io = get_file_io(); + let path = test_path("test_file_io_libhdfs_write_and_read"); + let output = file_io.new_output(&path).unwrap(); + output.write(Bytes::from_static(b"stale")).await.unwrap(); + + // The streaming writer overwrites the file. + let mut writer = output.writer().await.unwrap(); + writer.write(Bytes::from_static(b"01234")).await.unwrap(); + writer.write(Bytes::from_static(b"56789")).await.unwrap(); + writer.close().await.unwrap(); + + assert!(file_io.exists(&path).await.unwrap()); + let input = file_io.new_input(&path).unwrap(); + assert_eq!(input.metadata().await.unwrap().size, 10); + assert_eq!( + input.read().await.unwrap(), + Bytes::from_static(b"0123456789") + ); + let reader = input.reader().await.unwrap(); + assert_eq!( + reader.read(3..7).await.unwrap(), + Bytes::from_static(b"3456") + ); + } + + #[tokio::test] + async fn test_file_io_libhdfs_delete() { + let file_io = get_file_io(); + let dir = test_path("test_file_io_libhdfs_delete"); + let paths: Vec = (0..4).map(|i| format!("{dir}/file-{i}")).collect(); + for path in &paths { + file_io + .new_output(path) + .unwrap() + .write(Bytes::from_static(b"x")) + .await + .unwrap(); + } + + file_io.delete(&paths[0]).await.unwrap(); + file_io + .delete_stream(futures::stream::iter(paths[1..3].to_vec()).boxed()) + .await + .unwrap(); + for path in &paths[..3] { + assert!(!file_io.exists(path).await.unwrap(), "{path}"); + } + assert!(file_io.exists(&paths[3]).await.unwrap()); + file_io.delete_prefix(&dir).await.unwrap(); + assert!(!file_io.exists(&paths[3]).await.unwrap()); + } + + /// `dev/hdfs/client/core-site.xml` points `fs.defaultFS` at the fixture. + #[tokio::test] + async fn test_file_io_libhdfs_authority_less_path() { + let file_io = get_file_io(); + let name = normalize_test_name_with_parts!("test_file_io_libhdfs_authority_less_path"); + file_io + .new_output(format!("hdfs:///{name}")) + .unwrap() + .write(Bytes::from_static(b"x")) + .await + .unwrap(); + + let path = format!("{}/{name}", get_hdfs_endpoint()); + assert!(file_io.exists(&path).await.unwrap()); + file_io.delete(&path).await.unwrap(); + } +} diff --git a/dev/hdfs/client/core-site.xml b/dev/hdfs/client/core-site.xml new file mode 100644 index 0000000000..004e332b7d --- /dev/null +++ b/dev/hdfs/client/core-site.xml @@ -0,0 +1,32 @@ + + + + + + fs.defaultFS + hdfs://localhost:8020 + + + + dfs.client.use.datanode.hostname + true + + From cfad31217c2971151401410d831fd0dbdc3ab236 Mon Sep 17 00:00:00 2001 From: comphead Date: Thu, 22 Oct 2026 17:58:20 -0700 Subject: [PATCH 2/2] ci: build Windows without opendal-hdfs `hdrs` 0.3.4, the libhdfs binding behind `opendal-hdfs`, does not compile on Windows: it uses the Unix-only `libc::O_CLOEXEC` and assumes a 64-bit `tTime`. OpenDAL's own CI leaves `services-hdfs` out of its Windows build as well. --- .github/workflows/ci.yml | 11 ++++++++++- crates/storage/opendal/README.md | 2 +- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0c4d2f4d67..83921bf605 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -152,7 +152,16 @@ jobs: save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} - name: Build - run: make build + shell: bash + run: | + if [ "$RUNNER_OS" = "Windows" ]; then + # `hdrs`, the libhdfs binding behind `opendal-hdfs`, does not + # compile on Windows, so build every other feature there. + cargo build --all-targets --all-features --workspace --exclude iceberg-storage-opendal + cargo build --all-targets -p iceberg-storage-opendal --features opendal-all,opendal-hdfs-native + else + make build + fi # Checks each crate individually to catch missing feature declarations that # are otherwise masked by Cargo's feature unification across the workspace. diff --git a/crates/storage/opendal/README.md b/crates/storage/opendal/README.md index 3211ed2f3d..827375234d 100644 --- a/crates/storage/opendal/README.md +++ b/crates/storage/opendal/README.md @@ -41,7 +41,7 @@ You can enable all stable storage backends at once using the `opendal-all` featu > `opendal-hdfs-native` is experimental and not included in `opendal-all`. A portless path authority such as `hdfs://nameservice1/warehouse` must be declared in properties, e.g. `hdfs.name-node.nameservice1 = nn1:8020,nn2:8020`: unlike Hadoop, nameservices defined only in `$HADOOP_CONF_DIR` are not resolved, and a portless host is not dialed on the default port 8020. -> `opendal-hdfs` is experimental and not included in `opendal-all`. Use it through `OpenDalStorageFactory::Hdfs`, since `OpenDalResolvingStorageFactory` does not route to it. Hadoop's Java client reads its configuration from `CLASSPATH`, so nameservices, HA and Kerberos work as in Hadoop, and no `hdfs.*` or `hadoop.*` property applies. Building needs a JDK, and binaries need `$JAVA_HOME/lib/server` on the library path. Unless the process already runs a JVM, `CLASSPATH` must list the Hadoop configuration directory and each Hadoop jar, e.g. from `hadoop classpath --glob`. +> `opendal-hdfs` is experimental and not included in `opendal-all`. Use it through `OpenDalStorageFactory::Hdfs`, since `OpenDalResolvingStorageFactory` does not route to it. Hadoop's Java client reads its configuration from `CLASSPATH`, so nameservices, HA and Kerberos work as in Hadoop, and no `hdfs.*` or `hadoop.*` property applies. Building needs a JDK, and binaries need `$JAVA_HOME/lib/server` on the library path. It does not build on Windows, where its `hdrs` dependency does not compile. Unless the process already runs a JVM, `CLASSPATH` must list the Hadoop configuration directory and each Hadoop jar, e.g. from `hadoop classpath --glob`. ## Usage