feat(storage): Add HDFS support via opendal services-hdfs-native

crates/iceberg/public-api.txt

@@ -1072,6 +1072,10 @@ pub const iceberg::io::GCS_PROJECT_ID: &str
 pub const iceberg::io::GCS_SERVICE_HOST: &str
 pub const iceberg::io::GCS_TOKEN: &str
 pub const iceberg::io::GCS_USER_PROJECT: &str
+pub const iceberg::io::HDFS_HADOOP_CONF_PREFIX: &str
+pub const iceberg::io::HDFS_HOST: &str
+pub const iceberg::io::HDFS_NAME_NODE: &str
+pub const iceberg::io::HDFS_PORT: &str
 pub const iceberg::io::HF_ENDPOINT: &str
 pub const iceberg::io::HF_REVISION: &str
 pub const iceberg::io::HF_TOKEN: &str

crates/iceberg/src/io/storage/config/hdfs_native.rs

@@ -0,0 +1,45 @@
+// 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 configuration.
+
+/// HDFS NameNode RPC endpoint(s) as `host:port`, e.g. `hdfs://namenode:8020`;
+/// a comma-separated list enables HA failover. Serves authority-less paths
+/// (`hdfs:///path`) only; a path authority with a port is used as is. A
+/// portless authority (`hdfs://ns1/path`) must be a nameservice declared in
+/// properties: `hdfs.name-node.<nameservice>` takes the same list and is
+/// sugar for Hadoop's `dfs.ha.namenodes.<nameservice>` and
+/// `dfs.namenode.rpc-address.<nameservice>.<id>`, which are honored as well
+/// when passed through [`HDFS_HADOOP_CONF_PREFIX`]. Unlike Hadoop, an
+/// undeclared one is an error: it is neither resolved from
+/// `$HADOOP_CONF_DIR` nor dialed on the default port 8020.
+pub const HDFS_NAME_NODE: &str = "hdfs.name-node";
+/// NameNode host for authority-less paths, as in PyIceberg; paths that carry
+/// an authority ignore it, as they do there. Combined with [`HDFS_PORT`] into
+/// Hadoop's `fs.defaultFS`. PyIceberg's `hdfs.user` and `hdfs.kerberos_ticket`
+/// have no equivalent: the client reads `HADOOP_USER_NAME` and the default
+/// Kerberos credential cache.
+pub const HDFS_HOST: &str = "hdfs.host";
+/// NameNode port for [`HDFS_HOST`]; defaults to `8020` and has no effect
+/// without the host.
+pub const HDFS_PORT: &str = "hdfs.port";
+/// Prefix for properties forwarded to the HDFS client configuration with the
+/// prefix stripped: `hadoop.dfs.client.failover.random.order` is forwarded as
+/// `dfs.client.failover.random.order`. Forwarded values override those loaded
+/// from `$HADOOP_CONF_DIR`; `hadoop.fs.defaultFS` also serves authority-less
+/// paths.
+pub const HDFS_HADOOP_CONF_PREFIX: &str = "hadoop.";

crates/iceberg/src/io/storage/config/mod.rs

@@ -32,6 +32,7 @@
 
 mod azdls;
 mod gcs;
+mod hdfs_native;
 mod hf;
 mod oss;
 mod s3;
@@ -40,6 +41,7 @@ use std::collections::HashMap;
 
 pub use azdls::*;
 pub use gcs::*;
+pub use hdfs_native::*;
 pub use hf::*;
 pub use oss::*;
 pub use s3::*;

crates/storage/opendal/Cargo.toml

@@ -42,6 +42,11 @@ opendal-all = [
 opendal-azdls = ["opendal/services-azdls"]
 opendal-fs = ["opendal/services-fs"]
 opendal-gcs = ["opendal/services-gcs"]
+opendal-hdfs-native = [
+  "opendal/services-hdfs-native",
+  "dep:tokio",
+  "dep:tracing",
+]
 opendal-hf = ["opendal/services-hf"]
 opendal-memory = ["opendal/services-memory"]
 opendal-oss = ["opendal/services-oss"]
@@ -59,6 +64,8 @@ opendal = { workspace = true }
 reqsign-aws-v4 = { version = "3.0.0", optional = true }
 reqsign-core = { version = "3.0.0", optional = true }
 serde = { workspace = true }
+tokio = { workspace = true, optional = true }
+tracing = { workspace = true, optional = true }
 typetag = { workspace = true }
 url = { workspace = true }
 

crates/storage/opendal/README.md

@@ -32,11 +32,14 @@ OpenDAL-based storage backend implementations for [Apache Iceberg Rust](https://
 | Hugging Face         | `opendal-hf`     | ✅ Stable       | Hugging Face buckets and repositories         |
 | 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`) |
 
 You can enable all stable storage backends at once using the `opendal-all` feature flag.
 
 > Note that `opendal-oss` and `opendal-azdls` are currently experimental and not included in `opendal-all`.
 
+> `opendal-hdfs-native` is experimental and not included in `opendal-all`; see [HDFS](#hdfs).
+
 ## Usage
 
 Add the crate to your `Cargo.toml` with the feature flags for the backends you need:
@@ -82,3 +85,40 @@ async fn main() -> iceberg::Result<()> {
     Ok(())
 }
 ```
+
+## HDFS
+
+The `opendal-hdfs-native` feature adds an HDFS backend built on
+[`hdfs-native`](https://github.com/Kimahriman/hdfs-native), a native Rust client that needs no
+JVM. Enable it on its own, since it is not part of `opendal-all`:
+
+```toml
+iceberg-storage-opendal = { version = "x.y.z", features = ["opendal-hdfs-native"] }
+```
+
+Then pass `OpenDalStorageFactory::HdfsNative` to your catalog builder, or use
+`OpenDalResolvingStorageFactory`, which routes `hdfs://` paths to it. The backend reads these
+properties:
+
+| Property | Description |
+| --- | --- |
+| `hdfs.name-node` | NameNode `host:port` for authority-less paths (`hdfs:///warehouse/...`); a comma-separated list enables HA failover |
+| `hdfs.name-node.<nameservice>` | The NameNodes of a nameservice used as a path authority (`hdfs://nameservice1/warehouse/...`), comma-separated |
+| `hdfs.host`, `hdfs.port` | PyIceberg's keys for authority-less paths; the port defaults to `8020` |
+| `hadoop.*` | Forwarded to the HDFS client with the prefix stripped, e.g. `hadoop.dfs.client.failover.max.attempts` |
+
+A path with a `host:port` authority is always used as is. For a table under
+`hdfs://nameservice1/warehouse` on an HA cluster:
+
+```text
+hdfs.name-node.nameservice1 = nn1.example.com:8020,nn2.example.com:8020
+```
+
+The client also loads `core-site.xml` and `hdfs-site.xml` from `$HADOOP_CONF_DIR`, else
+`$HADOOP_HOME/etc/hadoop`, and takes the user from `HADOOP_USER_NAME`. Kerberos is used when
+`hadoop.security.authentication` is `kerberos` there, with the default credential cache
+(`kinit`); `libgssapi_krb5` is loaded at runtime. PyIceberg's `hdfs.user` and
+`hdfs.kerberos_ticket` are ignored with a warning.
+
+Unlike Hadoop, a portless authority is never resolved from `$HADOOP_CONF_DIR` or dialed on
+port 8020: declare it with `hdfs.name-node.<nameservice>`.

crates/storage/opendal/public-api.txt

@@ -8,6 +8,7 @@ pub iceberg_storage_opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc<ope
 pub iceberg_storage_opendal::OpenDalStorage::Gcs
 pub iceberg_storage_opendal::OpenDalStorage::Gcs::client_config: iceberg_storage_opendal::OpenDalClientConfig
 pub iceberg_storage_opendal::OpenDalStorage::Gcs::config: alloc::sync::Arc<opendal_service_gcs::config::GcsConfig>
+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
 pub iceberg_storage_opendal::OpenDalStorage::Hf::config: alloc::sync::Arc<opendal_service_hf::config::HfConfig>
@@ -47,6 +48,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::HdfsNative
 pub iceberg_storage_opendal::OpenDalStorageFactory::Hf
 pub iceberg_storage_opendal::OpenDalStorageFactory::Memory
 pub iceberg_storage_opendal::OpenDalStorageFactory::Oss
@@ -69,6 +71,15 @@ impl core::clone::Clone for iceberg_storage_opendal::CustomAwsCredentialLoader
 pub fn iceberg_storage_opendal::CustomAwsCredentialLoader::clone(&self) -> Self
 impl core::fmt::Debug for iceberg_storage_opendal::CustomAwsCredentialLoader
 pub fn iceberg_storage_opendal::CustomAwsCredentialLoader::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+pub struct iceberg_storage_opendal::HdfsNativeStorage
+impl core::clone::Clone for iceberg_storage_opendal::HdfsNativeStorage
+pub fn iceberg_storage_opendal::HdfsNativeStorage::clone(&self) -> iceberg_storage_opendal::HdfsNativeStorage
+impl core::fmt::Debug for iceberg_storage_opendal::HdfsNativeStorage
+pub fn iceberg_storage_opendal::HdfsNativeStorage::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+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<Self, <__D as serde_core::de::Deserializer>::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<alloc::string::String, alloc::string::String>) -> iceberg::error::Result<Self>