From 6679a1946c55a2643f54ffea1109768aea86c066 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Sat, 29 Aug 2026 17:27:30 +0300 Subject: [PATCH 1/7] feat(storage): Add HDFS support via opendal services-hdfs-native Adds an opendal-hdfs-native cargo feature plus OpenDalStorageFactory::Hdfs and OpenDalStorage::Hdfs variants in iceberg-storage-opendal, using OpenDAL's services-hdfs-native (pure-Rust HDFS RPC, no JNI/libhdfs). The NameNode for a path resolves as: the hdfs.name-node property when set (comma-separated endpoints enable HA failover), otherwise the path authority. hadoop.-prefixed properties are forwarded to the HDFS client configuration, overriding values loaded from $HADOOP_CONF_DIR. Operators are cached per effective NameNode since each holds live RPC connections. Revives and updates PR #2441 (by @jordepic) against the current storage layer and opendal 0.58, where name_node became mandatory and the comma list is the HA mechanism. Closes #2440 Co-authored-by: Jordan Epstein Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 10 + Cargo.lock | 286 ++++++++++++++++- crates/iceberg/public-api.txt | 27 ++ crates/iceberg/src/io/storage/config/hdfs.rs | 110 +++++++ crates/iceberg/src/io/storage/config/mod.rs | 2 + crates/storage/opendal/Cargo.toml | 1 + crates/storage/opendal/README.md | 3 +- crates/storage/opendal/public-api.txt | 4 + crates/storage/opendal/src/hdfs.rs | 289 +++++++++++++++++ crates/storage/opendal/src/lib.rs | 85 +++++ crates/storage/opendal/src/resolving.rs | 37 +++ .../opendal/tests/file_io_hdfs_test.rs | 292 ++++++++++++++++++ crates/test_utils/src/lib.rs | 9 + dev/docker-compose.yaml | 43 +++ dev/hdfs/core-site.xml | 27 ++ dev/hdfs/hdfs-site.xml | 35 +++ 16 files changed, 1247 insertions(+), 13 deletions(-) create mode 100644 crates/iceberg/src/io/storage/config/hdfs.rs create mode 100644 crates/storage/opendal/src/hdfs.rs create mode 100644 crates/storage/opendal/tests/file_io_hdfs_test.rs create mode 100644 dev/hdfs/core-site.xml create mode 100644 dev/hdfs/hdfs-site.xml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 394781eb0b..89a902b904 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -232,6 +232,16 @@ jobs: cargo test --no-fail-fast ${{ matrix.test-suite.args }} fi + # HDFS tests are `#[ignore]`d (docker fixture needs Linux host + # networking); opt them in here by test-name filter. + - name: Run HDFS integration tests + if: matrix.test-suite.name == 'default' + shell: bash + run: | + cargo nextest run ${{ matrix.test-suite.args }} \ + --run-ignored=only \ + -E 'test(file_io_hdfs)' + - name: Stop Docker containers if: always() && matrix.test-suite.name == 'default' run: make docker-down diff --git a/Cargo.lock b/Cargo.lock index 3802cc03ff..d143485a5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -25,10 +25,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if 1.0.4", - "cipher", + "cipher 0.4.4", "cpufeatures 0.2.17", ] +[[package]] +name = "aes" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35f0f96ce78e38c3dc6d8948aa8163d06385be74000f3c7a95bf1eef35d3ea32" +dependencies = [ + "cipher 0.5.2", + "cpubits", + "cpufeatures 0.3.0", +] + [[package]] name = "aes-gcm" version = "0.10.3" @@ -36,9 +47,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" dependencies = [ "aead", - "aes", - "cipher", - "ctr", + "aes 0.8.4", + "cipher 0.4.4", + "ctr 0.9.2", "ghash", "subtle", ] @@ -1193,6 +1204,15 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block-padding" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "710f1dd022ef4e93f8a438b4ba958de7f64308434fa6a87104481645cc30068b" +dependencies = [ + "hybrid-array", +] + [[package]] name = "bnum" version = "0.12.1" @@ -1335,7 +1355,16 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6" dependencies = [ - "cipher", + "cipher 0.4.4", +] + +[[package]] +name = "cbc" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce2dc9ee5f88d11e0beb842c88b33c8a5cf0d1329c4b19494af42b07dbfe8896" +dependencies = [ + "cipher 0.5.2", ] [[package]] @@ -1410,7 +1439,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ "crypto-common 0.1.7", - "inout", + "inout 0.1.4", +] + +[[package]] +name = "cipher" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8cf2a2c93cd704877c0858356ed03480ff301ee950b43f1cbe4573b088bfa6c" +dependencies = [ + "block-buffer 0.12.1", + "crypto-common 0.2.2", + "inout 0.2.2", ] [[package]] @@ -1642,6 +1682,12 @@ dependencies = [ "futures-io", ] +[[package]] +name = "cpubits" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15b85f9c39137c3a891689859392b1bd49812121d0d61c9caf00d46ed5ce06ae" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -1806,7 +1852,16 @@ version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" dependencies = [ - "cipher", + "cipher 0.4.4", +] + +[[package]] +name = "ctr" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baaca1c4b237092596f64d571e9db6ce4109c4ef9742e27590f1709594461f21" +dependencies = [ + "cipher 0.5.2", ] [[package]] @@ -2816,6 +2871,15 @@ dependencies = [ "syn", ] +[[package]] +name = "des" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "916a94e407b54f9034d71dd748234cd1e516ced6284009906ae246f177eafe5a" +dependencies = [ + "cipher 0.5.2", +] + [[package]] name = "diff" version = "0.1.13" @@ -2893,6 +2957,18 @@ dependencies = [ "const-random", ] +[[package]] +name = "dns-lookup" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e39034cee21a2f5bbb66ba0e3689819c4bb5d00382a282006e802a7ffa6c41d" +dependencies = [ + "cfg-if 1.0.4", + "libc", + "socket2 0.6.4", + "windows-sys 0.60.2", +] + [[package]] name = "dotenvy" version = "0.15.7" @@ -3319,6 +3395,34 @@ dependencies = [ "slab", ] +[[package]] +name = "g2gen" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5a7e0eb46f83a20260b850117d204366674e85d3a908d90865c78df9a6b1dfc" +dependencies = [ + "g2poly", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "g2p" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "539e2644c030d3bf4cd208cb842d2ce2f80e82e6e8472390bcef83ceba0d80ad" +dependencies = [ + "g2gen", + "g2poly", +] + +[[package]] +name = "g2poly" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "312d2295c7302019c395cfb90dacd00a82a2eabd700429bba9c7a3f38dbbe11b" + [[package]] name = "gearhash" version = "0.1.3" @@ -3500,6 +3604,47 @@ dependencies = [ "hashbrown 0.15.5", ] +[[package]] +name = "hdfs-native" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5cf4000cab03f95f56fac82175a2978c21517e6072e577ac8d402f00ce996997" +dependencies = [ + "aes 0.9.3", + "base64 0.22.1", + "bitflags 2.13.0", + "bumpalo", + "bytes", + "cbc 0.2.1", + "chrono", + "cipher 0.5.2", + "crc", + "ctr 0.10.1", + "des", + "dns-lookup", + "futures", + "g2p", + "hex", + "hmac 0.13.0", + "libc", + "libloading", + "log", + "md-5 0.11.0", + "num-traits", + "once_cell", + "prost", + "prost-types", + "rand 0.10.2", + "regex", + "roxmltree", + "socket2 0.6.4", + "thiserror 2.0.18", + "tokio", + "url", + "uuid", + "whoami 2.1.2", +] + [[package]] name = "heapify" version = "0.2.0" @@ -4228,10 +4373,20 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" dependencies = [ - "block-padding", + "block-padding 0.3.3", "generic-array", ] +[[package]] +name = "inout" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4250ce6452e92010fdf7268ccc5d14faa80bb12fc741938534c58f16804e03c7" +dependencies = [ + "block-padding 0.4.2", + "hybrid-array", +] + [[package]] name = "integer-encoding" version = "3.0.4" @@ -5173,6 +5328,7 @@ dependencies = [ "opendal-service-azdls", "opendal-service-fs", "opendal-service-gcs", + "opendal-service-hdfs-native", "opendal-service-hf", "opendal-service-oss", "opendal-service-s3", @@ -5327,6 +5483,20 @@ dependencies = [ "tokio", ] +[[package]] +name = "opendal-service-hdfs-native" +version = "0.58.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bdeb0c6fd99ffe3acf9c58f9664842ea549e4fd984c8202a5e656d18de46dae4" +dependencies = [ + "bytes", + "futures", + "hdfs-native", + "log", + "opendal-core", + "serde", +] + [[package]] name = "opendal-service-hf" version = "0.58.1" @@ -5652,8 +5822,8 @@ version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e847e2c91a18bfa887dd028ec33f2fe6f25db77db3619024764914affe8b69a6" dependencies = [ - "aes", - "cbc", + "aes 0.8.4", + "cbc 0.1.2", "der", "pbkdf2", "scrypt", @@ -5823,6 +5993,15 @@ dependencies = [ "syn", ] +[[package]] +name = "prost-types" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f94967dc7688f3054c7fac87473ffae4cc4c3904800e2d9f5b857246d8963b0a" +dependencies = [ + "prost", +] + [[package]] name = "psm" version = "0.1.31" @@ -6535,6 +6714,15 @@ dependencies = [ "byteorder", ] +[[package]] +name = "roxmltree" +version = "0.21.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1964b10c76125c36f8afe190065a4bf9a87bf324842c05701330bba9f1cacbb" +dependencies = [ + "memchr", +] + [[package]] name = "rsa" version = "0.9.10" @@ -6747,7 +6935,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" dependencies = [ - "cipher", + "cipher 0.4.4", ] [[package]] @@ -8800,6 +8988,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" +dependencies = [ + "windows-targets 0.53.5", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -8833,13 +9030,30 @@ dependencies = [ "windows_aarch64_gnullvm 0.52.6", "windows_aarch64_msvc 0.52.6", "windows_i686_gnu 0.52.6", - "windows_i686_gnullvm", + "windows_i686_gnullvm 0.52.6", "windows_i686_msvc 0.52.6", "windows_x86_64_gnu 0.52.6", "windows_x86_64_gnullvm 0.52.6", "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows-targets" +version = "0.53.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" +dependencies = [ + "windows-link", + "windows_aarch64_gnullvm 0.53.1", + "windows_aarch64_msvc 0.53.1", + "windows_i686_gnu 0.53.1", + "windows_i686_gnullvm 0.53.1", + "windows_i686_msvc 0.53.1", + "windows_x86_64_gnu 0.53.1", + "windows_x86_64_gnullvm 0.53.1", + "windows_x86_64_msvc 0.53.1", +] + [[package]] name = "windows-threading" version = "0.2.1" @@ -8861,6 +9075,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" + [[package]] name = "windows_aarch64_msvc" version = "0.48.5" @@ -8873,6 +9093,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_aarch64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" + [[package]] name = "windows_i686_gnu" version = "0.48.5" @@ -8885,12 +9111,24 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" +[[package]] +name = "windows_i686_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" + [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" + [[package]] name = "windows_i686_msvc" version = "0.48.5" @@ -8903,6 +9141,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_i686_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" + [[package]] name = "windows_x86_64_gnu" version = "0.48.5" @@ -8915,6 +9159,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" +[[package]] +name = "windows_x86_64_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" + [[package]] name = "windows_x86_64_gnullvm" version = "0.48.5" @@ -8927,6 +9177,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" + [[package]] name = "windows_x86_64_msvc" version = "0.48.5" @@ -8939,6 +9195,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "windows_x86_64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" + [[package]] name = "winnow" version = "0.7.15" diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 4aeddb665a..757fc4abb2 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -780,6 +780,28 @@ impl serde_core::ser::Serialize for iceberg::io::GcsConfig pub fn iceberg::io::GcsConfig::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::io::GcsConfig pub fn iceberg::io::GcsConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::io::HdfsConfig +pub iceberg::io::HdfsConfig::name_node: core::option::Option +pub iceberg::io::HdfsConfig::options: std::collections::hash::map::HashMap +impl core::clone::Clone for iceberg::io::HdfsConfig +pub fn iceberg::io::HdfsConfig::clone(&self) -> iceberg::io::HdfsConfig +impl core::cmp::Eq for iceberg::io::HdfsConfig +impl core::cmp::PartialEq for iceberg::io::HdfsConfig +pub fn iceberg::io::HdfsConfig::eq(&self, other: &iceberg::io::HdfsConfig) -> bool +impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::HdfsConfig +pub type iceberg::io::HdfsConfig::Error = iceberg::Error +pub fn iceberg::io::HdfsConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result +impl core::default::Default for iceberg::io::HdfsConfig +pub fn iceberg::io::HdfsConfig::default() -> iceberg::io::HdfsConfig +impl core::fmt::Debug for iceberg::io::HdfsConfig +pub fn iceberg::io::HdfsConfig::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg::io::HdfsConfig +impl iceberg::io::HdfsConfig +pub fn iceberg::io::HdfsConfig::builder() -> HdfsConfigBuilder<((), ())> +impl serde_core::ser::Serialize for iceberg::io::HdfsConfig +pub fn iceberg::io::HdfsConfig::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::io::HdfsConfig +pub fn iceberg::io::HdfsConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::io::HfConfig pub iceberg::io::HfConfig::endpoint: core::option::Option pub iceberg::io::HfConfig::revision: core::option::Option @@ -979,6 +1001,9 @@ pub fn iceberg::io::AzdlsConfig::try_from(config: &iceberg::io::StorageConfig) - impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::GcsConfig pub type iceberg::io::GcsConfig::Error = iceberg::Error pub fn iceberg::io::GcsConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result +impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::HdfsConfig +pub type iceberg::io::HdfsConfig::Error = iceberg::Error +pub fn iceberg::io::HdfsConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::HfConfig pub type iceberg::io::HfConfig::Error = iceberg::Error pub fn iceberg::io::HfConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result @@ -1015,6 +1040,8 @@ 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_NAME_NODE: &str pub const iceberg::io::HF_ENDPOINT: &str pub const iceberg::io::HF_REVISION: &str pub const iceberg::io::HF_TOKEN: &str diff --git a/crates/iceberg/src/io/storage/config/hdfs.rs b/crates/iceberg/src/io/storage/config/hdfs.rs new file mode 100644 index 0000000000..9c0d99734b --- /dev/null +++ b/crates/iceberg/src/io/storage/config/hdfs.rs @@ -0,0 +1,110 @@ +// 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. + +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; +use typed_builder::TypedBuilder; + +use super::StorageConfig; +use crate::Result; + +/// HDFS NameNode RPC endpoint(s), e.g. `hdfs://namenode:8020`; a +/// comma-separated list enables HA failover. When unset, the NameNode is +/// derived from the path authority. +pub const HDFS_NAME_NODE: &str = "hdfs.name-node"; +/// Prefix for properties forwarded to the HDFS client configuration, e.g. +/// `hadoop.dfs.client.failover.random.order`. Forwarded values (prefix +/// stripped) override those loaded from `$HADOOP_CONF_DIR`. +pub const HDFS_HADOOP_CONF_PREFIX: &str = "hadoop."; + +/// HDFS storage configuration. +/// +/// This struct contains all the configuration options for connecting to HDFS. +/// Use the builder pattern via `HdfsConfig::builder()` to construct instances. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, TypedBuilder)] +pub struct HdfsConfig { + /// NameNode endpoint(s); comma-separated for HA failover. + #[builder(default, setter(strip_option, into))] + pub name_node: Option, + /// Extra HDFS client configuration (the `hadoop.` prefix stripped). + #[builder(default)] + pub options: HashMap, +} + +impl TryFrom<&StorageConfig> for HdfsConfig { + type Error = crate::Error; + + fn try_from(config: &StorageConfig) -> Result { + let props = config.props(); + + let mut cfg = HdfsConfig::default(); + + if let Some(name_node) = props.get(HDFS_NAME_NODE) { + cfg.name_node = Some(name_node.clone()); + } + for (key, value) in props { + if let Some(stripped) = key.strip_prefix(HDFS_HADOOP_CONF_PREFIX) { + cfg.options.insert(stripped.to_string(), value.clone()); + } + } + + Ok(cfg) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_hdfs_config_builder() { + let cfg = HdfsConfig::builder() + .name_node("hdfs://nn1:8020,hdfs://nn2:8020") + .build(); + assert_eq!( + cfg.name_node.as_deref(), + Some("hdfs://nn1:8020,hdfs://nn2:8020") + ); + assert!(cfg.options.is_empty()); + } + + #[test] + fn test_hdfs_config_from_storage_config() { + let storage_config = StorageConfig::new() + .with_prop(HDFS_NAME_NODE, "hdfs://namenode:8020") + .with_prop("hadoop.dfs.client.failover.random.order", "true") + .with_prop("unrelated.key", "ignored"); + + let cfg = HdfsConfig::try_from(&storage_config).unwrap(); + assert_eq!(cfg.name_node.as_deref(), Some("hdfs://namenode:8020")); + assert_eq!( + cfg.options.get("dfs.client.failover.random.order"), + Some(&"true".to_string()) + ); + assert!(!cfg.options.contains_key("unrelated.key")); + } + + #[test] + fn test_hdfs_config_empty() { + let cfg = HdfsConfig::try_from(&StorageConfig::new()).unwrap(); + assert_eq!(cfg.name_node, None); + assert!(cfg.options.is_empty()); + } +} diff --git a/crates/iceberg/src/io/storage/config/mod.rs b/crates/iceberg/src/io/storage/config/mod.rs index d8d356de16..fe545d7a5b 100644 --- a/crates/iceberg/src/io/storage/config/mod.rs +++ b/crates/iceberg/src/io/storage/config/mod.rs @@ -32,6 +32,7 @@ mod azdls; mod gcs; +mod hdfs; mod hf; mod oss; mod s3; @@ -40,6 +41,7 @@ use std::collections::HashMap; pub use azdls::*; pub use gcs::*; +pub use hdfs::*; pub use hf::*; pub use oss::*; pub use s3::*; diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index e43e7845b3..de1ac7b143 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-native = ["opendal/services-hdfs-native"] opendal-hf = ["opendal/services-hf"] opendal-memory = ["opendal/services-memory"] opendal-oss = ["opendal/services-oss"] diff --git a/crates/storage/opendal/README.md b/crates/storage/opendal/README.md index 5db4293e67..45fa27e582 100644 --- a/crates/storage/opendal/README.md +++ b/crates/storage/opendal/README.md @@ -32,10 +32,11 @@ 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`. +> Note that `opendal-oss`, `opendal-azdls` and `opendal-hdfs-native` are currently experimental and not included in `opendal-all`. ## Usage diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index d8c4ecdb38..2a562eaaf3 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -6,6 +6,9 @@ pub iceberg_storage_opendal::OpenDalStorage::Azdls pub iceberg_storage_opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::Gcs pub iceberg_storage_opendal::OpenDalStorage::Gcs::config: alloc::sync::Arc +pub iceberg_storage_opendal::OpenDalStorage::Hdfs +pub iceberg_storage_opendal::OpenDalStorage::Hdfs::config: alloc::sync::Arc +pub iceberg_storage_opendal::OpenDalStorage::Hdfs::operators: alloc::sync::Arc>> pub iceberg_storage_opendal::OpenDalStorage::Hf pub iceberg_storage_opendal::OpenDalStorage::Hf::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::LocalFs @@ -39,6 +42,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::Hf pub iceberg_storage_opendal::OpenDalStorageFactory::Memory pub iceberg_storage_opendal::OpenDalStorageFactory::Oss diff --git a/crates/storage/opendal/src/hdfs.rs b/crates/storage/opendal/src/hdfs.rs new file mode 100644 index 0000000000..1c0dfd3da4 --- /dev/null +++ b/crates/storage/opendal/src/hdfs.rs @@ -0,0 +1,289 @@ +// 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-native` (pure Rust, no JNI). + +use std::collections::HashMap; +use std::sync::RwLock; + +use iceberg::io::{HDFS_HADOOP_CONF_PREFIX, HDFS_NAME_NODE}; +use iceberg::{Error, ErrorKind, Result}; +use opendal::Operator; +use opendal::services::HdfsNativeConfig; +use url::Url; + +use crate::utils::from_opendal_error; + +/// Parse iceberg properties to [`HdfsNativeConfig`]. +pub(crate) fn hdfs_config_parse(mut m: HashMap) -> Result { + let mut cfg = HdfsNativeConfig::default(); + + if let Some(name_node) = m.remove(HDFS_NAME_NODE) { + cfg.name_node = Some(name_node); + } + + let options: HashMap = m + .into_iter() + .filter_map(|(key, value)| { + key.strip_prefix(HDFS_HADOOP_CONF_PREFIX) + .map(|stripped| (stripped.to_string(), value)) + }) + .collect(); + if !options.is_empty() { + cfg.options = Some(options); + } + + Ok(cfg) +} + +/// Parse an HDFS path into `Some("hdfs://")` (`None` when +/// authority-less) and the relative path (no leading `/`, opendal style). +pub(crate) fn parse_hdfs_path(path: &str) -> Result<(Option, &str)> { + let url = Url::parse(path).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}: {e}"), + ) + })?; + if url.scheme() != "hdfs" { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}, expected scheme `hdfs://`"), + )); + } + + let name_node = url.host_str().filter(|h| !h.is_empty()).map(|host| { + url.port() + .map(|port| format!("hdfs://{host}:{port}")) + .unwrap_or_else(|| format!("hdfs://{host}")) + }); + + // `url.path()` borrows from `url` and can't be returned with the input's + // lifetime. Slice the path component out of the original input instead; + // it starts after the first `/` following the `hdfs://` prefix. Opendal + // paths must not start with `/` (`Deleter::delete` rejects them). + let after_scheme = &path["hdfs://".len()..]; + let rel = match after_scheme.find('/') { + Some(i) => after_scheme[i..].trim_start_matches('/'), + None => "", + }; + + Ok((name_node, rel)) +} + +/// Creates an operator for the path, cached per effective NameNode (the +/// configured `hdfs.name-node`, else the path authority) โ€” each operator +/// holds an HDFS client with live RPC connections. +pub(crate) fn hdfs_create_operator<'a>( + path: &'a str, + config: &HdfsNativeConfig, + operators: &RwLock>, +) -> Result<(Operator, &'a str)> { + let (authority_name_node, relative_path) = parse_hdfs_path(path)?; + + let name_node = match config.name_node.clone().or(authority_name_node) { + Some(name_node) => name_node, + None => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid hdfs path: {path}, authority-less paths require the `{HDFS_NAME_NODE}` property" + ), + )); + } + }; + + // Fast path: check read lock first. + { + let cache = operators + .read() + .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; + if let Some(op) = cache.get(&name_node) { + return Ok((op.clone(), relative_path)); + } + } + + // Slow path: build and insert under write lock, re-checking for a + // concurrent insert. + let mut cache = operators + .write() + .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; + let op = match cache.get(&name_node) { + Some(op) => op.clone(), + None => { + let op = hdfs_operator_build(config, &name_node)?; + cache.insert(name_node, op.clone()); + op + } + }; + + Ok((op, relative_path)) +} + +/// Build a new OpenDAL [`Operator`]: OpenDAL splits `name_node` on commas +/// into a synthetic HA name service; `$HADOOP_CONF_DIR` XML still merges in. +fn hdfs_operator_build(config: &HdfsNativeConfig, name_node: &str) -> Result { + let mut cfg = config.clone(); + cfg.name_node = Some(name_node.to_string()); + Operator::from_config(cfg).map_err(from_opendal_error) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_hdfs_config_parse_name_node_and_options() { + let props = HashMap::from([ + ( + HDFS_NAME_NODE.to_string(), + "hdfs://nn1:8020,hdfs://nn2:8020".to_string(), + ), + ( + "hadoop.dfs.client.failover.random.order".to_string(), + "true".to_string(), + ), + ("unrelated.key".to_string(), "ignored".to_string()), + ]); + + let cfg = hdfs_config_parse(props).unwrap(); + + assert_eq!( + cfg.name_node.as_deref(), + Some("hdfs://nn1:8020,hdfs://nn2:8020") + ); + let options = cfg.options.unwrap(); + assert_eq!( + options.get("dfs.client.failover.random.order"), + Some(&"true".to_string()) + ); + assert!(!options.contains_key("unrelated.key")); + } + + #[test] + fn test_hdfs_config_parse_empty() { + let cfg = hdfs_config_parse(HashMap::new()).unwrap(); + + assert_eq!(cfg.name_node, None); + assert_eq!(cfg.options, None); + } + + #[test] + fn test_parse_hdfs_path_with_authority_and_rel() { + let (nn, rel) = parse_hdfs_path("hdfs://nameservice1/a/b").unwrap(); + + assert_eq!(nn.as_deref(), Some("hdfs://nameservice1")); + assert_eq!(rel, "a/b"); + } + + #[test] + fn test_parse_hdfs_path_with_authority_and_port() { + let (nn, rel) = parse_hdfs_path("hdfs://nn:8020/foo").unwrap(); + + assert_eq!(nn.as_deref(), Some("hdfs://nn:8020")); + assert_eq!(rel, "foo"); + } + + #[test] + fn test_parse_hdfs_path_with_authority_no_path() { + let (nn, rel) = parse_hdfs_path("hdfs://nameservice1").unwrap(); + + assert_eq!(nn.as_deref(), Some("hdfs://nameservice1")); + assert_eq!(rel, ""); + } + + #[test] + fn test_parse_hdfs_path_with_authority_trailing_slash() { + let (nn, rel) = parse_hdfs_path("hdfs://nameservice1/").unwrap(); + + assert_eq!(nn.as_deref(), Some("hdfs://nameservice1")); + assert_eq!(rel, ""); + } + + #[test] + fn test_parse_hdfs_path_authority_less_returns_none() { + let (nn, rel) = parse_hdfs_path("hdfs:///a/b").unwrap(); + + assert_eq!(nn, None); + assert_eq!(rel, "a/b"); + } + + #[test] + fn test_parse_hdfs_path_wrong_scheme_errors() { + let err = parse_hdfs_path("file:///tmp/x").unwrap_err(); + + assert!(err.to_string().contains("expected scheme `hdfs://`")); + } + + #[test] + fn test_parse_hdfs_path_invalid_url_errors() { + let err = parse_hdfs_path("not-a-url").unwrap_err(); + + assert!(err.to_string().contains("Invalid hdfs path")); + } + + #[test] + fn test_hdfs_create_operator_configured_name_node_wins() { + let config = hdfs_config_parse(HashMap::from([( + HDFS_NAME_NODE.to_string(), + "hdfs://configured:8020".to_string(), + )])) + .unwrap(); + let operators = RwLock::new(HashMap::new()); + + let (_, rel) = + hdfs_create_operator("hdfs://from-path:9000/a/b", &config, &operators).unwrap(); + + assert_eq!(rel, "a/b"); + let cache = operators.read().unwrap(); + assert!(cache.contains_key("hdfs://configured:8020")); + assert!(!cache.contains_key("hdfs://from-path:9000")); + } + + #[test] + fn test_hdfs_create_operator_uses_path_authority() { + let config = HdfsNativeConfig::default(); + let operators = RwLock::new(HashMap::new()); + + let (_, rel) = hdfs_create_operator("hdfs://nn:8020/a/b", &config, &operators).unwrap(); + + assert_eq!(rel, "a/b"); + assert!(operators.read().unwrap().contains_key("hdfs://nn:8020")); + } + + #[test] + fn test_hdfs_create_operator_caches_per_name_node() { + let config = HdfsNativeConfig::default(); + let operators = RwLock::new(HashMap::new()); + + hdfs_create_operator("hdfs://nn1:8020/a", &config, &operators).unwrap(); + hdfs_create_operator("hdfs://nn1:8020/b", &config, &operators).unwrap(); + hdfs_create_operator("hdfs://nn2:8020/c", &config, &operators).unwrap(); + + assert_eq!(operators.read().unwrap().len(), 2); + } + + #[test] + fn test_hdfs_create_operator_authority_less_without_config_errors() { + let config = HdfsNativeConfig::default(); + let operators = RwLock::new(HashMap::new()); + + let err = hdfs_create_operator("hdfs:///a/b", &config, &operators).unwrap_err(); + + assert!(err.to_string().contains(HDFS_NAME_NODE)); + } +} diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index d02507e7b5..6626c356e1 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -74,6 +74,16 @@ cfg_if! { } } +cfg_if! { + if #[cfg(feature = "opendal-hdfs-native")] { + mod hdfs; + use std::sync::RwLock; + + use hdfs::*; + use opendal::services::HdfsNativeConfig; + } +} + cfg_if! { if #[cfg(feature = "opendal-memory")] { mod memory; @@ -122,6 +132,9 @@ pub enum OpenDalStorageFactory { /// GCS storage factory. #[cfg(feature = "opendal-gcs")] Gcs, + /// HDFS storage factory. + #[cfg(feature = "opendal-hdfs-native")] + Hdfs, /// OSS storage factory. #[cfg(feature = "opendal-oss")] Oss, @@ -155,6 +168,11 @@ impl StorageFactory for OpenDalStorageFactory { OpenDalStorageFactory::Gcs => Ok(Arc::new(OpenDalStorage::Gcs { config: gcs_config_parse(config.props().clone())?.into(), })), + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorageFactory::Hdfs => Ok(Arc::new(OpenDalStorage::Hdfs { + config: hdfs_config_parse(config.props().clone())?.into(), + operators: Arc::new(RwLock::new(HashMap::new())), + })), #[cfg(feature = "opendal-oss")] OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::Oss { config: oss_config_parse(config.props().clone())?.into(), @@ -175,6 +193,7 @@ impl StorageFactory for OpenDalStorageFactory { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs-native"), ))] _ => Err(Error::new( ErrorKind::FeatureUnsupported, @@ -217,6 +236,18 @@ pub enum OpenDalStorage { /// GCS configuration. config: Arc, }, + /// HDFS storage variant. + /// + /// The NameNode is taken from the `hdfs.name-node` property when set + /// (comma-separated endpoints enable HA failover), else the path authority. + #[cfg(feature = "opendal-hdfs-native")] + Hdfs { + /// HDFS configuration. + config: Arc, + /// Operator cache keyed by effective NameNode. + #[serde(skip, default)] + operators: Arc>>, + }, /// OSS storage variant. #[cfg(feature = "opendal-oss")] Oss { @@ -322,6 +353,10 @@ impl OpenDalStorage { )); } } + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorage::Hdfs { config, operators } => { + hdfs_create_operator(path, config, operators)? + } #[cfg(feature = "opendal-oss")] OpenDalStorage::Oss { config } => { let op = oss_config_build(config, path)?; @@ -346,6 +381,7 @@ impl OpenDalStorage { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs-native"), ))] _ => { return Err(Error::new( @@ -434,6 +470,11 @@ impl OpenDalStorage { )) } } + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorage::Hdfs { .. } => { + let (_, relative_path) = parse_hdfs_path(path)?; + Ok(relative_path) + } #[cfg(feature = "opendal-oss")] OpenDalStorage::Oss { .. } => { let url = url::Url::parse(path)?; @@ -474,6 +515,7 @@ impl OpenDalStorage { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs-native"), ))] _ => Err(Error::new( ErrorKind::FeatureUnsupported, @@ -750,6 +792,49 @@ mod tests { ); } + #[cfg(feature = "opendal-hdfs-native")] + fn hdfs_test_storage() -> OpenDalStorage { + OpenDalStorage::Hdfs { + config: Arc::new(HdfsNativeConfig::default()), + operators: Arc::new(RwLock::new(HashMap::new())), + } + } + + #[cfg(feature = "opendal-hdfs-native")] + #[test] + fn test_relativize_path_hdfs() { + let storage = hdfs_test_storage(); + + assert_eq!( + storage + .relativize_path("hdfs://nameservice1/a/b.parquet") + .unwrap(), + "a/b.parquet" + ); + assert_eq!( + storage + .relativize_path("hdfs://nn:8020/warehouse/db/t") + .unwrap(), + "warehouse/db/t" + ); + } + + #[cfg(feature = "opendal-hdfs-native")] + #[test] + fn test_relativize_path_hdfs_authority_less() { + let storage = hdfs_test_storage(); + + assert_eq!(storage.relativize_path("hdfs:///a/b").unwrap(), "a/b"); + } + + #[cfg(feature = "opendal-hdfs-native")] + #[test] + fn test_relativize_path_hdfs_wrong_scheme_errors() { + let storage = hdfs_test_storage(); + + assert!(storage.relativize_path("s3://bucket/x").is_err()); + } + #[cfg(feature = "opendal-azdls")] #[test] fn test_relativize_path_azdls() { diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 86993220a8..8f7ad22288 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -51,6 +51,7 @@ pub const SCHEME_ABFS: &str = "abfs"; pub const SCHEME_WASBS: &str = "wasbs"; pub const SCHEME_WASB: &str = "wasb"; pub const SCHEME_HF: &str = "hf"; +pub const SCHEME_HDFS: &str = "hdfs"; /// Parse a URL scheme string. fn parse_scheme(scheme: &str) -> Result<&'static str> { @@ -62,6 +63,7 @@ fn parse_scheme(scheme: &str) -> Result<&'static str> { SCHEME_OSS => Ok("oss"), SCHEME_ABFSS | SCHEME_ABFS | SCHEME_WASBS | SCHEME_WASB => Ok("azdls"), SCHEME_HF => Ok("hf"), + SCHEME_HDFS => Ok("hdfs"), s => Err(Error::new( ErrorKind::FeatureUnsupported, format!("Unsupported storage scheme: {s}"), @@ -127,6 +129,14 @@ fn build_storage_for_scheme( config: Arc::new(config), }) } + #[cfg(feature = "opendal-hdfs-native")] + "hdfs" => { + let config = crate::hdfs::hdfs_config_parse(props.clone())?; + Ok(OpenDalStorage::Hdfs { + config: Arc::new(config), + operators: Arc::new(RwLock::new(HashMap::new())), + }) + } unsupported => Err(Error::new( ErrorKind::FeatureUnsupported, format!("Unsupported storage scheme: {unsupported}"), @@ -353,6 +363,33 @@ mod tests { assert!(Arc::ptr_eq(&a, &c), "s3 and s3n should share one instance"); } + #[cfg(feature = "opendal-hdfs-native")] + #[test] + fn test_resolve_hdfs_returns_hdfs_variant() { + let storage = empty_resolving_storage(); + + let resolved = storage.resolve("hdfs://nameservice1/a/b").unwrap(); + + assert!( + matches!(&*resolved, OpenDalStorage::Hdfs { .. }), + "expected Hdfs variant, got {resolved:?}" + ); + } + + #[cfg(feature = "opendal-hdfs-native")] + #[test] + fn test_resolve_hdfs_distinct_authorities_share_instance() { + let storage = empty_resolving_storage(); + + let a = storage.resolve("hdfs://ns1/a").unwrap(); + let b = storage.resolve("hdfs://ns2/b").unwrap(); + + assert!( + Arc::ptr_eq(&a, &b), + "different authorities should share the OpenDalStorage::Hdfs instance (operator cache is internal)" + ); + } + #[cfg(feature = "opendal-azdls")] #[test] fn test_resolve_azdls_aliases_share_instance() { diff --git a/crates/storage/opendal/tests/file_io_hdfs_test.rs b/crates/storage/opendal/tests/file_io_hdfs_test.rs new file mode 100644 index 0000000000..0f7bd00584 --- /dev/null +++ b/crates/storage/opendal/tests/file_io_hdfs_test.rs @@ -0,0 +1,292 @@ +// 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-native`. +//! +//! These tests need the `hdfs-namenode`/`hdfs-datanode` services from +//! `dev/docker-compose.yaml` and are `#[ignore]`d (host networking is +//! Linux-only); CI opts in via `cargo nextest run --run-ignored=only`. + +#[cfg(feature = "opendal-hdfs-native")] +mod tests { + use std::sync::Arc; + + use bytes::Bytes; + use futures::StreamExt; + use iceberg::io::{FileIO, FileIOBuilder, HDFS_NAME_NODE}; + use iceberg_storage_opendal::{OpenDalResolvingStorageFactory, 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] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_exists() { + let file_io = get_file_io(); + + let absent = test_path("test_file_io_hdfs_exists_absent"); + assert!(!file_io.exists(&absent).await.unwrap()); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_write_and_read() { + let file_io = get_file_io(); + let path = test_path("test_file_io_hdfs_write_and_read"); + let _ = file_io.delete(&path).await; + + let output = file_io.new_output(&path).unwrap(); + output + .write(Bytes::from_static(b"hello hdfs")) + .await + .unwrap(); + + assert!(file_io.exists(&path).await.unwrap()); + let input = file_io.new_input(&path).unwrap(); + assert_eq!( + input.read().await.unwrap(), + Bytes::from_static(b"hello hdfs") + ); + } + + /// The HA flow: table locations carry a logical authority while + /// `hdfs.name-node` carries the (comma-separated) endpoints; it wins. + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_configured_name_node() { + set_up(); + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Hdfs)) + .with_prop(HDFS_NAME_NODE, get_hdfs_endpoint()) + .build(); + + // The path authority is a logical name; the configured NameNode wins. + let path = format!( + "hdfs://logical-nameservice/{}", + normalize_test_name_with_parts!("test_file_io_hdfs_configured_name_node") + ); + let _ = file_io.delete(&path).await; + + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"via configured name node")) + .await + .unwrap(); + + assert!(file_io.exists(&path).await.unwrap()); + assert_eq!( + file_io.new_input(&path).unwrap().read().await.unwrap(), + Bytes::from_static(b"via configured name node") + ); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_overwrite() { + let file_io = get_file_io(); + let path = test_path("test_file_io_hdfs_overwrite"); + let _ = file_io.delete(&path).await; + + for content in [b"first".as_slice(), b"second, longer".as_slice()] { + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(content)) + .await + .unwrap(); + } + + assert_eq!( + file_io.new_input(&path).unwrap().read().await.unwrap(), + Bytes::from_static(b"second, longer") + ); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_delete_stream() { + let file_io = get_file_io(); + + let paths: Vec = (0..5) + .map(|i| format!("{}/file-{i}", test_path("test_file_io_hdfs_delete_stream"))) + .collect(); + for path in &paths { + let _ = file_io.delete(path).await; + file_io + .new_output(path) + .unwrap() + .write("delete-me".into()) + .await + .unwrap(); + assert!(file_io.exists(path).await.unwrap()); + } + + let stream = futures::stream::iter(paths.clone()).boxed(); + file_io.delete_stream(stream).await.unwrap(); + + for path in &paths { + assert!(!file_io.exists(path).await.unwrap()); + } + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_delete_stream_empty() { + let file_io = get_file_io(); + let stream = futures::stream::empty().boxed(); + file_io.delete_stream(stream).await.unwrap(); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_resolving_storage() { + set_up(); + let file_io = FileIOBuilder::new(Arc::new(OpenDalResolvingStorageFactory::new())).build(); + let path = test_path("test_file_io_hdfs_resolving_storage"); + let _ = file_io.delete(&path).await; + + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"resolving")) + .await + .unwrap(); + + assert_eq!( + file_io.new_input(&path).unwrap().read().await.unwrap(), + Bytes::from_static(b"resolving") + ); + + file_io.delete(&path).await.unwrap(); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_metadata() { + let file_io = get_file_io(); + let path = test_path("test_file_io_hdfs_metadata"); + let _ = file_io.delete(&path).await; + let content = Bytes::from_static(b"0123456789"); + + file_io + .new_output(&path) + .unwrap() + .write(content.clone()) + .await + .unwrap(); + + let metadata = file_io.new_input(&path).unwrap().metadata().await.unwrap(); + assert_eq!(metadata.size, content.len() as u64); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_delete() { + let file_io = get_file_io(); + let path = test_path("test_file_io_hdfs_delete"); + + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"x")) + .await + .unwrap(); + assert!(file_io.exists(&path).await.unwrap()); + + file_io.delete(&path).await.unwrap(); + assert!(!file_io.exists(&path).await.unwrap()); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_delete_prefix() { + let file_io = get_file_io(); + let dir = test_path("test_file_io_hdfs_delete_prefix"); + let _ = file_io.delete_prefix(&dir).await; + + for i in 0..3 { + let path = format!("{dir}/file_{i}"); + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from(format!("payload {i}"))) + .await + .unwrap(); + } + assert!(file_io.exists(&format!("{dir}/file_0")).await.unwrap()); + + file_io.delete_prefix(&dir).await.unwrap(); + + for i in 0..3 { + assert!(!file_io.exists(&format!("{dir}/file_{i}")).await.unwrap()); + } + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_reader_range() { + let file_io = get_file_io(); + let path = test_path("test_file_io_hdfs_reader_range"); + let _ = file_io.delete(&path).await; + let content = Bytes::from_static(b"abcdefghij"); + + file_io + .new_output(&path) + .unwrap() + .write(content.clone()) + .await + .unwrap(); + + let reader = file_io.new_input(&path).unwrap().reader().await.unwrap(); + assert_eq!( + reader.read(0..5).await.unwrap(), + Bytes::from_static(b"abcde") + ); + assert_eq!( + reader.read(5..10).await.unwrap(), + Bytes::from_static(b"fghij") + ); + } + + #[tokio::test] + #[ignore = "Linux-only: HDFS docker fixture uses host networking"] + async fn test_file_io_hdfs_streaming_writer() { + let file_io = get_file_io(); + let path = test_path("test_file_io_hdfs_streaming_writer"); + let _ = file_io.delete(&path).await; + + let output = file_io.new_output(&path).unwrap(); + let mut writer = output.writer().await.unwrap(); + writer.write(Bytes::from_static(b"part1 ")).await.unwrap(); + writer.write(Bytes::from_static(b"part2")).await.unwrap(); + writer.close().await.unwrap(); + + let read = file_io.new_input(&path).unwrap().read().await.unwrap(); + assert_eq!(read, Bytes::from_static(b"part1 part2")); + } +} diff --git a/crates/test_utils/src/lib.rs b/crates/test_utils/src/lib.rs index e44d96c385..f027df12e9 100644 --- a/crates/test_utils/src/lib.rs +++ b/crates/test_utils/src/lib.rs @@ -42,6 +42,7 @@ mod common { pub const ENV_HMS_ENDPOINT: &str = "ICEBERG_TEST_HMS_ENDPOINT"; pub const ENV_GLUE_ENDPOINT: &str = "ICEBERG_TEST_GLUE_ENDPOINT"; pub const ENV_GCS_ENDPOINT: &str = "ICEBERG_TEST_GCS_ENDPOINT"; + pub const ENV_HDFS_ENDPOINT: &str = "ICEBERG_TEST_HDFS_ENDPOINT"; // Default ports matching dev/docker-compose.yaml pub const DEFAULT_MINIO_PORT: u16 = 9000; @@ -49,6 +50,7 @@ mod common { pub const DEFAULT_HMS_PORT: u16 = 9083; pub const DEFAULT_GLUE_PORT: u16 = 5001; pub const DEFAULT_GCS_PORT: u16 = 4443; + pub const DEFAULT_HDFS_NN_PORT: u16 = 8020; /// Returns the MinIO S3-compatible endpoint. /// Checks ICEBERG_TEST_MINIO_ENDPOINT env var, otherwise returns localhost default. @@ -84,6 +86,13 @@ mod common { .unwrap_or_else(|_| format!("http://localhost:{DEFAULT_GCS_PORT}")) } + /// Returns the HDFS NameNode endpoint (e.g. `hdfs://localhost:8020`). + /// Checks ICEBERG_TEST_HDFS_ENDPOINT env var, otherwise returns localhost default. + pub fn get_hdfs_endpoint() -> String { + std::env::var(ENV_HDFS_ENDPOINT) + .unwrap_or_else(|_| format!("hdfs://localhost:{DEFAULT_HDFS_NN_PORT}")) + } + /// Helper to clean up a namespace and its tables before a test runs. /// This handles the case where previous test runs left data in the persistent database. pub async fn cleanup_namespace(catalog: &C, ns: &NamespaceIdent) { diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml index 9d288580df..6416e8983f 100644 --- a/dev/docker-compose.yaml +++ b/dev/docker-compose.yaml @@ -147,6 +147,49 @@ services: timeout: 5s retries: 5 + # ============================================================================= + # HDFS - single-node NameNode + DataNode for HDFS tests + # ============================================================================= + # hdfs-native connects to DataNodes by their registered IP โ€” unroutable on + # a docker bridge, hence host networking (Linux-only; tests are `#[ignore]`d). + hdfs-namenode: + image: apache/hadoop:3.5.0 + network_mode: "host" + command: ["hdfs", "namenode"] + environment: + ENSURE_NAMENODE_DIR: "/tmp/hadoop-root/dfs/name" + extra_hosts: + - "docker-desktop:127.0.0.1" + volumes: + - ./hdfs/core-site.xml:/opt/hadoop/etc/hadoop/core-site.xml:ro + - ./hdfs/hdfs-site.xml:/opt/hadoop/etc/hadoop/hdfs-site.xml:ro + healthcheck: + test: ["CMD-SHELL", "hdfs dfsadmin -safemode get | grep -q OFF"] + interval: 5s + timeout: 15s + retries: 30 + start_period: 30s + + hdfs-datanode: + image: apache/hadoop:3.5.0 + network_mode: "host" + command: ["hdfs", "datanode"] + depends_on: + hdfs-namenode: + condition: service_healthy + extra_hosts: + - "docker-desktop:127.0.0.1" + volumes: + - ./hdfs/core-site.xml:/opt/hadoop/etc/hadoop/core-site.xml:ro + - ./hdfs/hdfs-site.xml:/opt/hadoop/etc/hadoop/hdfs-site.xml:ro + # Healthy only once registered with the NameNode โ€” writes fail until then. + healthcheck: + test: ["CMD-SHELL", "hdfs dfsadmin -report | grep -q 'Live datanodes (1)'"] + interval: 5s + timeout: 15s + retries: 30 + start_period: 10s + # ============================================================================= # Fake GCS Server - GCS emulator for GCS tests # ============================================================================= diff --git a/dev/hdfs/core-site.xml b/dev/hdfs/core-site.xml new file mode 100644 index 0000000000..3c57323d7b --- /dev/null +++ b/dev/hdfs/core-site.xml @@ -0,0 +1,27 @@ + + + + + fs.defaultFS + hdfs://localhost:8020 + + + hadoop.http.staticuser.user + root + + diff --git a/dev/hdfs/hdfs-site.xml b/dev/hdfs/hdfs-site.xml new file mode 100644 index 0000000000..5eb2d198f1 --- /dev/null +++ b/dev/hdfs/hdfs-site.xml @@ -0,0 +1,35 @@ + + + + + dfs.replication + 1 + + + dfs.permissions.enabled + false + + + dfs.namenode.name.dir + file:///tmp/hadoop-root/dfs/name + + + dfs.datanode.data.dir + file:///tmp/hadoop-root/dfs/data + + From 2b0ca954e65a57be5f4969c5442ed7557c2f6008 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Mon, 31 Aug 2026 12:17:15 +0300 Subject: [PATCH 2/7] =?UTF-8?q?refactor:=20address=20review=20=E2=80=94=20?= =?UTF-8?q?Properties=20derive=20for=20HdfsConfig,=20drop=20HDFS=20CI=20sp?= =?UTF-8?q?ecial-casing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - HdfsConfig is now pub(crate) and parsed via #[derive(Properties)] (key/prefix attributes) instead of hand-written TryFrom + TypedBuilder. - HDFS integration tests are no longer #[ignore]d and the dedicated CI step is gone; they run under the default nextest invocation since make docker-up already starts the fixture and the Tests job is Linux-only. Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 10 --- crates/iceberg/public-api.txt | 25 ------ crates/iceberg/src/io/storage/config/hdfs.rs | 90 +++++++------------ .../opendal/tests/file_io_hdfs_test.rs | 16 +--- dev/docker-compose.yaml | 2 +- 5 files changed, 33 insertions(+), 110 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 89a902b904..394781eb0b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -232,16 +232,6 @@ jobs: cargo test --no-fail-fast ${{ matrix.test-suite.args }} fi - # HDFS tests are `#[ignore]`d (docker fixture needs Linux host - # networking); opt them in here by test-name filter. - - name: Run HDFS integration tests - if: matrix.test-suite.name == 'default' - shell: bash - run: | - cargo nextest run ${{ matrix.test-suite.args }} \ - --run-ignored=only \ - -E 'test(file_io_hdfs)' - - name: Stop Docker containers if: always() && matrix.test-suite.name == 'default' run: make docker-down diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 757fc4abb2..97673c15fb 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -780,28 +780,6 @@ impl serde_core::ser::Serialize for iceberg::io::GcsConfig pub fn iceberg::io::GcsConfig::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::io::GcsConfig pub fn iceberg::io::GcsConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> -pub struct iceberg::io::HdfsConfig -pub iceberg::io::HdfsConfig::name_node: core::option::Option -pub iceberg::io::HdfsConfig::options: std::collections::hash::map::HashMap -impl core::clone::Clone for iceberg::io::HdfsConfig -pub fn iceberg::io::HdfsConfig::clone(&self) -> iceberg::io::HdfsConfig -impl core::cmp::Eq for iceberg::io::HdfsConfig -impl core::cmp::PartialEq for iceberg::io::HdfsConfig -pub fn iceberg::io::HdfsConfig::eq(&self, other: &iceberg::io::HdfsConfig) -> bool -impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::HdfsConfig -pub type iceberg::io::HdfsConfig::Error = iceberg::Error -pub fn iceberg::io::HdfsConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result -impl core::default::Default for iceberg::io::HdfsConfig -pub fn iceberg::io::HdfsConfig::default() -> iceberg::io::HdfsConfig -impl core::fmt::Debug for iceberg::io::HdfsConfig -pub fn iceberg::io::HdfsConfig::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl core::marker::StructuralPartialEq for iceberg::io::HdfsConfig -impl iceberg::io::HdfsConfig -pub fn iceberg::io::HdfsConfig::builder() -> HdfsConfigBuilder<((), ())> -impl serde_core::ser::Serialize for iceberg::io::HdfsConfig -pub fn iceberg::io::HdfsConfig::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::io::HdfsConfig -pub fn iceberg::io::HdfsConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::io::HfConfig pub iceberg::io::HfConfig::endpoint: core::option::Option pub iceberg::io::HfConfig::revision: core::option::Option @@ -1001,9 +979,6 @@ pub fn iceberg::io::AzdlsConfig::try_from(config: &iceberg::io::StorageConfig) - impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::GcsConfig pub type iceberg::io::GcsConfig::Error = iceberg::Error pub fn iceberg::io::GcsConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result -impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::HdfsConfig -pub type iceberg::io::HdfsConfig::Error = iceberg::Error -pub fn iceberg::io::HdfsConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::HfConfig pub type iceberg::io::HfConfig::Error = iceberg::Error pub fn iceberg::io::HfConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result diff --git a/crates/iceberg/src/io/storage/config/hdfs.rs b/crates/iceberg/src/io/storage/config/hdfs.rs index 9c0d99734b..e2d468dec1 100644 --- a/crates/iceberg/src/io/storage/config/hdfs.rs +++ b/crates/iceberg/src/io/storage/config/hdfs.rs @@ -19,11 +19,7 @@ use std::collections::HashMap; -use serde::{Deserialize, Serialize}; -use typed_builder::TypedBuilder; - -use super::StorageConfig; -use crate::Result; +use iceberg_property_macro::Properties; /// HDFS NameNode RPC endpoint(s), e.g. `hdfs://namenode:8020`; a /// comma-separated list enables HA failover. When unset, the NameNode is @@ -35,38 +31,17 @@ pub const HDFS_NAME_NODE: &str = "hdfs.name-node"; pub const HDFS_HADOOP_CONF_PREFIX: &str = "hadoop."; /// HDFS storage configuration. -/// -/// This struct contains all the configuration options for connecting to HDFS. -/// Use the builder pattern via `HdfsConfig::builder()` to construct instances. -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, TypedBuilder)] -pub struct HdfsConfig { +// No in-crate consumer yet: `iceberg-storage-opendal` parses the raw +// properties itself and only shares the key constants above. +#[allow(dead_code)] +#[derive(Debug, Properties)] +pub(crate) struct HdfsConfig { /// NameNode endpoint(s); comma-separated for HA failover. - #[builder(default, setter(strip_option, into))] - pub name_node: Option, + #[property(key = HDFS_NAME_NODE, default = None, getter)] + name_node: Option, /// Extra HDFS client configuration (the `hadoop.` prefix stripped). - #[builder(default)] - pub options: HashMap, -} - -impl TryFrom<&StorageConfig> for HdfsConfig { - type Error = crate::Error; - - fn try_from(config: &StorageConfig) -> Result { - let props = config.props(); - - let mut cfg = HdfsConfig::default(); - - if let Some(name_node) = props.get(HDFS_NAME_NODE) { - cfg.name_node = Some(name_node.clone()); - } - for (key, value) in props { - if let Some(stripped) = key.strip_prefix(HDFS_HADOOP_CONF_PREFIX) { - cfg.options.insert(stripped.to_string(), value.clone()); - } - } - - Ok(cfg) - } + #[property(prefix = HDFS_HADOOP_CONF_PREFIX, getter)] + options: HashMap, } #[cfg(test)] @@ -74,37 +49,32 @@ mod tests { use super::*; #[test] - fn test_hdfs_config_builder() { - let cfg = HdfsConfig::builder() - .name_node("hdfs://nn1:8020,hdfs://nn2:8020") - .build(); - assert_eq!( - cfg.name_node.as_deref(), - Some("hdfs://nn1:8020,hdfs://nn2:8020") - ); - assert!(cfg.options.is_empty()); - } - - #[test] - fn test_hdfs_config_from_storage_config() { - let storage_config = StorageConfig::new() - .with_prop(HDFS_NAME_NODE, "hdfs://namenode:8020") - .with_prop("hadoop.dfs.client.failover.random.order", "true") - .with_prop("unrelated.key", "ignored"); - - let cfg = HdfsConfig::try_from(&storage_config).unwrap(); - assert_eq!(cfg.name_node.as_deref(), Some("hdfs://namenode:8020")); + fn test_hdfs_config_from_properties() { + let props = HashMap::from([ + ( + HDFS_NAME_NODE.to_string(), + "hdfs://namenode:8020".to_string(), + ), + ( + "hadoop.dfs.client.failover.random.order".to_string(), + "true".to_string(), + ), + ("unrelated.key".to_string(), "ignored".to_string()), + ]); + + let cfg = HdfsConfig::from_properties(&props).unwrap(); + assert_eq!(cfg.name_node().as_deref(), Some("hdfs://namenode:8020")); assert_eq!( - cfg.options.get("dfs.client.failover.random.order"), + cfg.options().get("dfs.client.failover.random.order"), Some(&"true".to_string()) ); - assert!(!cfg.options.contains_key("unrelated.key")); + assert!(!cfg.options().contains_key("unrelated.key")); } #[test] fn test_hdfs_config_empty() { - let cfg = HdfsConfig::try_from(&StorageConfig::new()).unwrap(); - assert_eq!(cfg.name_node, None); - assert!(cfg.options.is_empty()); + let cfg = HdfsConfig::from_properties(&HashMap::new()).unwrap(); + assert_eq!(cfg.name_node().as_deref(), None); + assert!(cfg.options().is_empty()); } } diff --git a/crates/storage/opendal/tests/file_io_hdfs_test.rs b/crates/storage/opendal/tests/file_io_hdfs_test.rs index 0f7bd00584..f072a9c476 100644 --- a/crates/storage/opendal/tests/file_io_hdfs_test.rs +++ b/crates/storage/opendal/tests/file_io_hdfs_test.rs @@ -18,8 +18,8 @@ //! Integration tests for HDFS FileIO via OpenDAL `services-hdfs-native`. //! //! These tests need the `hdfs-namenode`/`hdfs-datanode` services from -//! `dev/docker-compose.yaml` and are `#[ignore]`d (host networking is -//! Linux-only); CI opts in via `cargo nextest run --run-ignored=only`. +//! `dev/docker-compose.yaml` (started by `make docker-up`); the fixture +//! uses host networking, which needs Linux or a recent Docker runtime. #[cfg(feature = "opendal-hdfs-native")] mod tests { @@ -45,7 +45,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_exists() { let file_io = get_file_io(); @@ -54,7 +53,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_write_and_read() { let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_write_and_read"); @@ -77,7 +75,6 @@ mod tests { /// The HA flow: table locations carry a logical authority while /// `hdfs.name-node` carries the (comma-separated) endpoints; it wins. #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_configured_name_node() { set_up(); let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Hdfs)) @@ -106,7 +103,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_overwrite() { let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_overwrite"); @@ -128,7 +124,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_delete_stream() { let file_io = get_file_io(); @@ -155,7 +150,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_delete_stream_empty() { let file_io = get_file_io(); let stream = futures::stream::empty().boxed(); @@ -163,7 +157,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_resolving_storage() { set_up(); let file_io = FileIOBuilder::new(Arc::new(OpenDalResolvingStorageFactory::new())).build(); @@ -186,7 +179,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_metadata() { let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_metadata"); @@ -205,7 +197,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_delete() { let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_delete"); @@ -223,7 +214,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_delete_prefix() { let file_io = get_file_io(); let dir = test_path("test_file_io_hdfs_delete_prefix"); @@ -248,7 +238,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_reader_range() { let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_reader_range"); @@ -274,7 +263,6 @@ mod tests { } #[tokio::test] - #[ignore = "Linux-only: HDFS docker fixture uses host networking"] async fn test_file_io_hdfs_streaming_writer() { let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_streaming_writer"); diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml index 6416e8983f..485adb900a 100644 --- a/dev/docker-compose.yaml +++ b/dev/docker-compose.yaml @@ -151,7 +151,7 @@ services: # HDFS - single-node NameNode + DataNode for HDFS tests # ============================================================================= # hdfs-native connects to DataNodes by their registered IP โ€” unroutable on - # a docker bridge, hence host networking (Linux-only; tests are `#[ignore]`d). + # a docker bridge, hence host networking (needs Linux or a recent runtime). hdfs-namenode: image: apache/hadoop:3.5.0 network_mode: "host" From 45bd68a8d3ab21f867e07453c389be51416ec962 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Mon, 31 Aug 2026 12:23:03 +0300 Subject: [PATCH 3/7] fix: reject non-hierarchical hdfs URLs; key delete_stream batches by effective NameNode - Url::parse accepts non-hierarchical forms like `hdfs:x`; the byte-7 slice then panicked. Require the literal `hdfs://` prefix instead. - batch_key_for_path grouped by URL host only, so NameNodes differing by port shared one deleter; key by the effective NameNode (configured hdfs.name-node, else authority incl. port), matching the operator cache. Co-Authored-By: Claude Opus 5 --- crates/storage/opendal/src/hdfs.rs | 66 ++++++++++++++++++++++++++++-- crates/storage/opendal/src/lib.rs | 4 ++ 2 files changed, 67 insertions(+), 3 deletions(-) diff --git a/crates/storage/opendal/src/hdfs.rs b/crates/storage/opendal/src/hdfs.rs index 1c0dfd3da4..9abfa70251 100644 --- a/crates/storage/opendal/src/hdfs.rs +++ b/crates/storage/opendal/src/hdfs.rs @@ -59,12 +59,14 @@ pub(crate) fn parse_hdfs_path(path: &str) -> Result<(Option, &str)> { format!("Invalid hdfs path: {path}: {e}"), ) })?; - if url.scheme() != "hdfs" { + // Non-special schemes parse even without `//` (e.g. `hdfs:x` is a valid + // non-hierarchical URL), so require the literal prefix before slicing. + let (Some(after_scheme), "hdfs") = (path.strip_prefix("hdfs://"), url.scheme()) else { return Err(Error::new( ErrorKind::DataInvalid, format!("Invalid hdfs path: {path}, expected scheme `hdfs://`"), )); - } + }; let name_node = url.host_str().filter(|h| !h.is_empty()).map(|host| { url.port() @@ -76,7 +78,6 @@ pub(crate) fn parse_hdfs_path(path: &str) -> Result<(Option, &str)> { // lifetime. Slice the path component out of the original input instead; // it starts after the first `/` following the `hdfs://` prefix. Opendal // paths must not start with `/` (`Deleter::delete` rejects them). - let after_scheme = &path["hdfs://".len()..]; let rel = match after_scheme.find('/') { Some(i) => after_scheme[i..].trim_start_matches('/'), None => "", @@ -134,6 +135,21 @@ pub(crate) fn hdfs_create_operator<'a>( Ok((op, relative_path)) } +/// Returns the `delete_stream` grouping key for a path: the effective +/// NameNode, mirroring the operator-cache key so paths that resolve to +/// different operators never share a deleter. +pub(crate) fn hdfs_batch_key(config: &HdfsNativeConfig, path: &str) -> String { + config + .name_node + .clone() + .or_else(|| { + parse_hdfs_path(path) + .ok() + .and_then(|(name_node, _)| name_node) + }) + .unwrap_or_default() +} + /// Build a new OpenDAL [`Operator`]: OpenDAL splits `name_node` on commas /// into a synthetic HA name service; `$HADOOP_CONF_DIR` XML still merges in. fn hdfs_operator_build(config: &HdfsNativeConfig, name_node: &str) -> Result { @@ -236,6 +252,50 @@ mod tests { assert!(err.to_string().contains("Invalid hdfs path")); } + #[test] + fn test_parse_hdfs_path_non_hierarchical_errors() { + // `hdfs:x` parses as a valid non-hierarchical URL; it must be + // rejected rather than panic on slicing. + for path in ["hdfs:x", "hdfs:/x", "hdfs:"] { + let err = parse_hdfs_path(path).unwrap_err(); + assert!(err.to_string().contains("expected scheme `hdfs://`")); + } + } + + #[test] + fn test_hdfs_batch_key_distinguishes_ports() { + let config = HdfsNativeConfig::default(); + + assert_eq!( + hdfs_batch_key(&config, "hdfs://namenode:8020/a"), + "hdfs://namenode:8020" + ); + assert_eq!( + hdfs_batch_key(&config, "hdfs://namenode:9000/b"), + "hdfs://namenode:9000" + ); + } + + #[test] + fn test_hdfs_batch_key_configured_name_node_wins() { + let config = hdfs_config_parse(HashMap::from([( + HDFS_NAME_NODE.to_string(), + "hdfs://nn1:8020,hdfs://nn2:8020".to_string(), + )])) + .unwrap(); + + // All paths group under the configured NameNode, matching the + // single cached operator they resolve to. + assert_eq!( + hdfs_batch_key(&config, "hdfs://ns-a/x"), + "hdfs://nn1:8020,hdfs://nn2:8020" + ); + assert_eq!( + hdfs_batch_key(&config, "hdfs:///y"), + "hdfs://nn1:8020,hdfs://nn2:8020" + ); + } + #[test] fn test_hdfs_create_operator_configured_name_node_wins() { let config = hdfs_config_parse(HashMap::from([( diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 6626c356e1..9e01793077 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -413,6 +413,10 @@ impl OpenDalStorage { match self { #[cfg(feature = "opendal-hf")] 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-native")] + OpenDalStorage::Hdfs { config, .. } => hdfs_batch_key(config, path), _ => url::Url::parse(path) .ok() .and_then(|u| u.host_str().map(|s| s.to_string())) From a4178fe023ab5e07f1dcda4b088cd72075e010de Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Sat, 5 Sep 2026 12:30:47 +0300 Subject: [PATCH 4/7] refactor(storage): rename hdfs module to hdfs_native, drop unused HdfsConfig MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review: the opendal module, its functions and the storage/factory variants are now hdfs_native-prefixed (leaving room for a libhdfs-backed variant, see #1130), and the never-consumed HdfsConfig struct is removed from the core crate โ€” config/hdfs.rs keeps only the property constants that iceberg-storage-opendal uses. Co-Authored-By: Claude Fable 5.1 --- crates/iceberg/src/io/storage/config/hdfs.rs | 53 ----------- crates/storage/opendal/public-api.txt | 8 +- .../opendal/src/{hdfs.rs => hdfs_native.rs} | 93 ++++++++++--------- crates/storage/opendal/src/lib.rs | 38 ++++---- crates/storage/opendal/src/resolving.rs | 14 +-- .../opendal/tests/file_io_hdfs_test.rs | 4 +- 6 files changed, 79 insertions(+), 131 deletions(-) rename crates/storage/opendal/src/{hdfs.rs => hdfs_native.rs} (73%) diff --git a/crates/iceberg/src/io/storage/config/hdfs.rs b/crates/iceberg/src/io/storage/config/hdfs.rs index e2d468dec1..626f23ffb5 100644 --- a/crates/iceberg/src/io/storage/config/hdfs.rs +++ b/crates/iceberg/src/io/storage/config/hdfs.rs @@ -17,10 +17,6 @@ //! HDFS storage configuration. -use std::collections::HashMap; - -use iceberg_property_macro::Properties; - /// HDFS NameNode RPC endpoint(s), e.g. `hdfs://namenode:8020`; a /// comma-separated list enables HA failover. When unset, the NameNode is /// derived from the path authority. @@ -29,52 +25,3 @@ pub const HDFS_NAME_NODE: &str = "hdfs.name-node"; /// `hadoop.dfs.client.failover.random.order`. Forwarded values (prefix /// stripped) override those loaded from `$HADOOP_CONF_DIR`. pub const HDFS_HADOOP_CONF_PREFIX: &str = "hadoop."; - -/// HDFS storage configuration. -// No in-crate consumer yet: `iceberg-storage-opendal` parses the raw -// properties itself and only shares the key constants above. -#[allow(dead_code)] -#[derive(Debug, Properties)] -pub(crate) struct HdfsConfig { - /// NameNode endpoint(s); comma-separated for HA failover. - #[property(key = HDFS_NAME_NODE, default = None, getter)] - name_node: Option, - /// Extra HDFS client configuration (the `hadoop.` prefix stripped). - #[property(prefix = HDFS_HADOOP_CONF_PREFIX, getter)] - options: HashMap, -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_hdfs_config_from_properties() { - let props = HashMap::from([ - ( - HDFS_NAME_NODE.to_string(), - "hdfs://namenode:8020".to_string(), - ), - ( - "hadoop.dfs.client.failover.random.order".to_string(), - "true".to_string(), - ), - ("unrelated.key".to_string(), "ignored".to_string()), - ]); - - let cfg = HdfsConfig::from_properties(&props).unwrap(); - assert_eq!(cfg.name_node().as_deref(), Some("hdfs://namenode:8020")); - assert_eq!( - cfg.options().get("dfs.client.failover.random.order"), - Some(&"true".to_string()) - ); - assert!(!cfg.options().contains_key("unrelated.key")); - } - - #[test] - fn test_hdfs_config_empty() { - let cfg = HdfsConfig::from_properties(&HashMap::new()).unwrap(); - assert_eq!(cfg.name_node().as_deref(), None); - assert!(cfg.options().is_empty()); - } -} diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index 2a562eaaf3..58bd9ee6b7 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -6,9 +6,9 @@ pub iceberg_storage_opendal::OpenDalStorage::Azdls pub iceberg_storage_opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::Gcs pub iceberg_storage_opendal::OpenDalStorage::Gcs::config: alloc::sync::Arc -pub iceberg_storage_opendal::OpenDalStorage::Hdfs -pub iceberg_storage_opendal::OpenDalStorage::Hdfs::config: alloc::sync::Arc -pub iceberg_storage_opendal::OpenDalStorage::Hdfs::operators: alloc::sync::Arc>> +pub iceberg_storage_opendal::OpenDalStorage::HdfsNative +pub iceberg_storage_opendal::OpenDalStorage::HdfsNative::config: alloc::sync::Arc +pub iceberg_storage_opendal::OpenDalStorage::HdfsNative::operators: alloc::sync::Arc>> pub iceberg_storage_opendal::OpenDalStorage::Hf pub iceberg_storage_opendal::OpenDalStorage::Hf::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::LocalFs @@ -42,7 +42,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 pub iceberg_storage_opendal::OpenDalStorageFactory::Oss diff --git a/crates/storage/opendal/src/hdfs.rs b/crates/storage/opendal/src/hdfs_native.rs similarity index 73% rename from crates/storage/opendal/src/hdfs.rs rename to crates/storage/opendal/src/hdfs_native.rs index 9abfa70251..9de9ecab5f 100644 --- a/crates/storage/opendal/src/hdfs.rs +++ b/crates/storage/opendal/src/hdfs_native.rs @@ -29,7 +29,7 @@ use url::Url; use crate::utils::from_opendal_error; /// Parse iceberg properties to [`HdfsNativeConfig`]. -pub(crate) fn hdfs_config_parse(mut m: HashMap) -> Result { +pub(crate) fn hdfs_native_config_parse(mut m: HashMap) -> Result { let mut cfg = HdfsNativeConfig::default(); if let Some(name_node) = m.remove(HDFS_NAME_NODE) { @@ -52,7 +52,7 @@ pub(crate) fn hdfs_config_parse(mut m: HashMap) -> Result")` (`None` when /// authority-less) and the relative path (no leading `/`, opendal style). -pub(crate) fn parse_hdfs_path(path: &str) -> Result<(Option, &str)> { +pub(crate) fn hdfs_native_parse_path(path: &str) -> Result<(Option, &str)> { let url = Url::parse(path).map_err(|e| { Error::new( ErrorKind::DataInvalid, @@ -89,12 +89,12 @@ pub(crate) fn parse_hdfs_path(path: &str) -> Result<(Option, &str)> { /// Creates an operator for the path, cached per effective NameNode (the /// configured `hdfs.name-node`, else the path authority) โ€” each operator /// holds an HDFS client with live RPC connections. -pub(crate) fn hdfs_create_operator<'a>( +pub(crate) fn hdfs_native_create_operator<'a>( path: &'a str, config: &HdfsNativeConfig, operators: &RwLock>, ) -> Result<(Operator, &'a str)> { - let (authority_name_node, relative_path) = parse_hdfs_path(path)?; + let (authority_name_node, relative_path) = hdfs_native_parse_path(path)?; let name_node = match config.name_node.clone().or(authority_name_node) { Some(name_node) => name_node, @@ -126,7 +126,7 @@ pub(crate) fn hdfs_create_operator<'a>( let op = match cache.get(&name_node) { Some(op) => op.clone(), None => { - let op = hdfs_operator_build(config, &name_node)?; + let op = hdfs_native_operator_build(config, &name_node)?; cache.insert(name_node, op.clone()); op } @@ -138,12 +138,12 @@ pub(crate) fn hdfs_create_operator<'a>( /// Returns the `delete_stream` grouping key for a path: the effective /// NameNode, mirroring the operator-cache key so paths that resolve to /// different operators never share a deleter. -pub(crate) fn hdfs_batch_key(config: &HdfsNativeConfig, path: &str) -> String { +pub(crate) fn hdfs_native_batch_key(config: &HdfsNativeConfig, path: &str) -> String { config .name_node .clone() .or_else(|| { - parse_hdfs_path(path) + hdfs_native_parse_path(path) .ok() .and_then(|(name_node, _)| name_node) }) @@ -152,7 +152,7 @@ pub(crate) fn hdfs_batch_key(config: &HdfsNativeConfig, path: &str) -> String { /// Build a new OpenDAL [`Operator`]: OpenDAL splits `name_node` on commas /// into a synthetic HA name service; `$HADOOP_CONF_DIR` XML still merges in. -fn hdfs_operator_build(config: &HdfsNativeConfig, name_node: &str) -> Result { +fn hdfs_native_operator_build(config: &HdfsNativeConfig, name_node: &str) -> Result { let mut cfg = config.clone(); cfg.name_node = Some(name_node.to_string()); Operator::from_config(cfg).map_err(from_opendal_error) @@ -163,7 +163,7 @@ mod tests { use super::*; #[test] - fn test_hdfs_config_parse_name_node_and_options() { + fn test_hdfs_native_config_parse_name_node_and_options() { let props = HashMap::from([ ( HDFS_NAME_NODE.to_string(), @@ -176,7 +176,7 @@ mod tests { ("unrelated.key".to_string(), "ignored".to_string()), ]); - let cfg = hdfs_config_parse(props).unwrap(); + let cfg = hdfs_native_config_parse(props).unwrap(); assert_eq!( cfg.name_node.as_deref(), @@ -191,94 +191,94 @@ mod tests { } #[test] - fn test_hdfs_config_parse_empty() { - let cfg = hdfs_config_parse(HashMap::new()).unwrap(); + fn test_hdfs_native_config_parse_empty() { + let cfg = hdfs_native_config_parse(HashMap::new()).unwrap(); assert_eq!(cfg.name_node, None); assert_eq!(cfg.options, None); } #[test] - fn test_parse_hdfs_path_with_authority_and_rel() { - let (nn, rel) = parse_hdfs_path("hdfs://nameservice1/a/b").unwrap(); + fn test_hdfs_native_parse_path_with_authority_and_rel() { + let (nn, rel) = hdfs_native_parse_path("hdfs://nameservice1/a/b").unwrap(); assert_eq!(nn.as_deref(), Some("hdfs://nameservice1")); assert_eq!(rel, "a/b"); } #[test] - fn test_parse_hdfs_path_with_authority_and_port() { - let (nn, rel) = parse_hdfs_path("hdfs://nn:8020/foo").unwrap(); + fn test_hdfs_native_parse_path_with_authority_and_port() { + let (nn, rel) = hdfs_native_parse_path("hdfs://nn:8020/foo").unwrap(); assert_eq!(nn.as_deref(), Some("hdfs://nn:8020")); assert_eq!(rel, "foo"); } #[test] - fn test_parse_hdfs_path_with_authority_no_path() { - let (nn, rel) = parse_hdfs_path("hdfs://nameservice1").unwrap(); + fn test_hdfs_native_parse_path_with_authority_no_path() { + let (nn, rel) = hdfs_native_parse_path("hdfs://nameservice1").unwrap(); assert_eq!(nn.as_deref(), Some("hdfs://nameservice1")); assert_eq!(rel, ""); } #[test] - fn test_parse_hdfs_path_with_authority_trailing_slash() { - let (nn, rel) = parse_hdfs_path("hdfs://nameservice1/").unwrap(); + fn test_hdfs_native_parse_path_with_authority_trailing_slash() { + let (nn, rel) = hdfs_native_parse_path("hdfs://nameservice1/").unwrap(); assert_eq!(nn.as_deref(), Some("hdfs://nameservice1")); assert_eq!(rel, ""); } #[test] - fn test_parse_hdfs_path_authority_less_returns_none() { - let (nn, rel) = parse_hdfs_path("hdfs:///a/b").unwrap(); + fn test_hdfs_native_parse_path_authority_less_returns_none() { + let (nn, rel) = hdfs_native_parse_path("hdfs:///a/b").unwrap(); assert_eq!(nn, None); assert_eq!(rel, "a/b"); } #[test] - fn test_parse_hdfs_path_wrong_scheme_errors() { - let err = parse_hdfs_path("file:///tmp/x").unwrap_err(); + fn test_hdfs_native_parse_path_wrong_scheme_errors() { + let err = hdfs_native_parse_path("file:///tmp/x").unwrap_err(); assert!(err.to_string().contains("expected scheme `hdfs://`")); } #[test] - fn test_parse_hdfs_path_invalid_url_errors() { - let err = parse_hdfs_path("not-a-url").unwrap_err(); + fn test_hdfs_native_parse_path_invalid_url_errors() { + let err = hdfs_native_parse_path("not-a-url").unwrap_err(); assert!(err.to_string().contains("Invalid hdfs path")); } #[test] - fn test_parse_hdfs_path_non_hierarchical_errors() { + fn test_hdfs_native_parse_path_non_hierarchical_errors() { // `hdfs:x` parses as a valid non-hierarchical URL; it must be // rejected rather than panic on slicing. for path in ["hdfs:x", "hdfs:/x", "hdfs:"] { - let err = parse_hdfs_path(path).unwrap_err(); + let err = hdfs_native_parse_path(path).unwrap_err(); assert!(err.to_string().contains("expected scheme `hdfs://`")); } } #[test] - fn test_hdfs_batch_key_distinguishes_ports() { + fn test_hdfs_native_batch_key_distinguishes_ports() { let config = HdfsNativeConfig::default(); assert_eq!( - hdfs_batch_key(&config, "hdfs://namenode:8020/a"), + hdfs_native_batch_key(&config, "hdfs://namenode:8020/a"), "hdfs://namenode:8020" ); assert_eq!( - hdfs_batch_key(&config, "hdfs://namenode:9000/b"), + hdfs_native_batch_key(&config, "hdfs://namenode:9000/b"), "hdfs://namenode:9000" ); } #[test] - fn test_hdfs_batch_key_configured_name_node_wins() { - let config = hdfs_config_parse(HashMap::from([( + fn test_hdfs_native_batch_key_configured_name_node_wins() { + let config = hdfs_native_config_parse(HashMap::from([( HDFS_NAME_NODE.to_string(), "hdfs://nn1:8020,hdfs://nn2:8020".to_string(), )])) @@ -287,18 +287,18 @@ mod tests { // All paths group under the configured NameNode, matching the // single cached operator they resolve to. assert_eq!( - hdfs_batch_key(&config, "hdfs://ns-a/x"), + hdfs_native_batch_key(&config, "hdfs://ns-a/x"), "hdfs://nn1:8020,hdfs://nn2:8020" ); assert_eq!( - hdfs_batch_key(&config, "hdfs:///y"), + hdfs_native_batch_key(&config, "hdfs:///y"), "hdfs://nn1:8020,hdfs://nn2:8020" ); } #[test] - fn test_hdfs_create_operator_configured_name_node_wins() { - let config = hdfs_config_parse(HashMap::from([( + fn test_hdfs_native_create_operator_configured_name_node_wins() { + let config = hdfs_native_config_parse(HashMap::from([( HDFS_NAME_NODE.to_string(), "hdfs://configured:8020".to_string(), )])) @@ -306,7 +306,7 @@ mod tests { let operators = RwLock::new(HashMap::new()); let (_, rel) = - hdfs_create_operator("hdfs://from-path:9000/a/b", &config, &operators).unwrap(); + hdfs_native_create_operator("hdfs://from-path:9000/a/b", &config, &operators).unwrap(); assert_eq!(rel, "a/b"); let cache = operators.read().unwrap(); @@ -315,34 +315,35 @@ mod tests { } #[test] - fn test_hdfs_create_operator_uses_path_authority() { + fn test_hdfs_native_create_operator_uses_path_authority() { let config = HdfsNativeConfig::default(); let operators = RwLock::new(HashMap::new()); - let (_, rel) = hdfs_create_operator("hdfs://nn:8020/a/b", &config, &operators).unwrap(); + let (_, rel) = + hdfs_native_create_operator("hdfs://nn:8020/a/b", &config, &operators).unwrap(); assert_eq!(rel, "a/b"); assert!(operators.read().unwrap().contains_key("hdfs://nn:8020")); } #[test] - fn test_hdfs_create_operator_caches_per_name_node() { + fn test_hdfs_native_create_operator_caches_per_name_node() { let config = HdfsNativeConfig::default(); let operators = RwLock::new(HashMap::new()); - hdfs_create_operator("hdfs://nn1:8020/a", &config, &operators).unwrap(); - hdfs_create_operator("hdfs://nn1:8020/b", &config, &operators).unwrap(); - hdfs_create_operator("hdfs://nn2:8020/c", &config, &operators).unwrap(); + hdfs_native_create_operator("hdfs://nn1:8020/a", &config, &operators).unwrap(); + hdfs_native_create_operator("hdfs://nn1:8020/b", &config, &operators).unwrap(); + hdfs_native_create_operator("hdfs://nn2:8020/c", &config, &operators).unwrap(); assert_eq!(operators.read().unwrap().len(), 2); } #[test] - fn test_hdfs_create_operator_authority_less_without_config_errors() { + fn test_hdfs_native_create_operator_authority_less_without_config_errors() { let config = HdfsNativeConfig::default(); let operators = RwLock::new(HashMap::new()); - let err = hdfs_create_operator("hdfs:///a/b", &config, &operators).unwrap_err(); + let err = hdfs_native_create_operator("hdfs:///a/b", &config, &operators).unwrap_err(); assert!(err.to_string().contains(HDFS_NAME_NODE)); } diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 0defd6dea9..d3bec9d23f 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -76,10 +76,10 @@ cfg_if! { cfg_if! { if #[cfg(feature = "opendal-hdfs-native")] { - mod hdfs; + mod hdfs_native; use std::sync::RwLock; - use hdfs::*; + use hdfs_native::*; use opendal::services::HdfsNativeConfig; } } @@ -147,7 +147,7 @@ pub enum OpenDalStorageFactory { Gcs, /// HDFS storage factory. #[cfg(feature = "opendal-hdfs-native")] - Hdfs, + HdfsNative, /// OSS storage factory. #[cfg(feature = "opendal-oss")] Oss, @@ -195,8 +195,8 @@ impl StorageFactory for OpenDalStorageFactory { config: gcs_config_parse(config.props().clone())?.into(), })), #[cfg(feature = "opendal-hdfs-native")] - OpenDalStorageFactory::Hdfs => Ok(Arc::new(OpenDalStorage::Hdfs { - config: hdfs_config_parse(config.props().clone())?.into(), + OpenDalStorageFactory::HdfsNative => Ok(Arc::new(OpenDalStorage::HdfsNative { + config: hdfs_native_config_parse(config.props().clone())?.into(), operators: Arc::new(RwLock::new(HashMap::new())), })), #[cfg(feature = "opendal-oss")] @@ -267,7 +267,7 @@ pub enum OpenDalStorage { /// The NameNode is taken from the `hdfs.name-node` property when set /// (comma-separated endpoints enable HA failover), else the path authority. #[cfg(feature = "opendal-hdfs-native")] - Hdfs { + HdfsNative { /// HDFS configuration. config: Arc, /// Operator cache keyed by effective NameNode. @@ -380,8 +380,8 @@ impl OpenDalStorage { } } #[cfg(feature = "opendal-hdfs-native")] - OpenDalStorage::Hdfs { config, operators } => { - hdfs_create_operator(path, config, operators)? + OpenDalStorage::HdfsNative { config, operators } => { + hdfs_native_create_operator(path, config, operators)? } #[cfg(feature = "opendal-oss")] OpenDalStorage::Oss { config } => { @@ -442,7 +442,7 @@ impl OpenDalStorage { // The URL host alone would merge distinct NameNodes that differ // only by port; key by the effective NameNode instead. #[cfg(feature = "opendal-hdfs-native")] - OpenDalStorage::Hdfs { config, .. } => hdfs_batch_key(config, path), + OpenDalStorage::HdfsNative { config, .. } => hdfs_native_batch_key(config, path), _ => url::Url::parse(path) .ok() .and_then(|u| u.host_str().map(|s| s.to_string())) @@ -501,8 +501,8 @@ impl OpenDalStorage { } } #[cfg(feature = "opendal-hdfs-native")] - OpenDalStorage::Hdfs { .. } => { - let (_, relative_path) = parse_hdfs_path(path)?; + OpenDalStorage::HdfsNative { .. } => { + let (_, relative_path) = hdfs_native_parse_path(path)?; Ok(relative_path) } #[cfg(feature = "opendal-oss")] @@ -854,8 +854,8 @@ mod tests { } #[cfg(feature = "opendal-hdfs-native")] - fn hdfs_test_storage() -> OpenDalStorage { - OpenDalStorage::Hdfs { + fn hdfs_native_test_storage() -> OpenDalStorage { + OpenDalStorage::HdfsNative { config: Arc::new(HdfsNativeConfig::default()), operators: Arc::new(RwLock::new(HashMap::new())), } @@ -863,8 +863,8 @@ mod tests { #[cfg(feature = "opendal-hdfs-native")] #[test] - fn test_relativize_path_hdfs() { - let storage = hdfs_test_storage(); + fn test_relativize_path_hdfs_native() { + let storage = hdfs_native_test_storage(); assert_eq!( storage @@ -882,16 +882,16 @@ mod tests { #[cfg(feature = "opendal-hdfs-native")] #[test] - fn test_relativize_path_hdfs_authority_less() { - let storage = hdfs_test_storage(); + fn test_relativize_path_hdfs_native_authority_less() { + let storage = hdfs_native_test_storage(); assert_eq!(storage.relativize_path("hdfs:///a/b").unwrap(), "a/b"); } #[cfg(feature = "opendal-hdfs-native")] #[test] - fn test_relativize_path_hdfs_wrong_scheme_errors() { - let storage = hdfs_test_storage(); + fn test_relativize_path_hdfs_native_wrong_scheme_errors() { + let storage = hdfs_native_test_storage(); assert!(storage.relativize_path("s3://bucket/x").is_err()); } diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 8a15ebd41f..b3af99928e 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -131,8 +131,8 @@ fn build_storage_for_scheme( } #[cfg(feature = "opendal-hdfs-native")] "hdfs" => { - let config = crate::hdfs::hdfs_config_parse(props.clone())?; - Ok(OpenDalStorage::Hdfs { + let config = crate::hdfs_native::hdfs_native_config_parse(props.clone())?; + Ok(OpenDalStorage::HdfsNative { config: Arc::new(config), operators: Arc::new(RwLock::new(HashMap::new())), }) @@ -405,20 +405,20 @@ mod tests { #[cfg(feature = "opendal-hdfs-native")] #[test] - fn test_resolve_hdfs_returns_hdfs_variant() { + fn test_resolve_hdfs_native_returns_hdfs_variant() { let storage = empty_resolving_storage(); let resolved = storage.resolve("hdfs://nameservice1/a/b").unwrap(); assert!( - matches!(&*resolved, OpenDalStorage::Hdfs { .. }), - "expected Hdfs variant, got {resolved:?}" + matches!(&*resolved, OpenDalStorage::HdfsNative { .. }), + "expected HdfsNative variant, got {resolved:?}" ); } #[cfg(feature = "opendal-hdfs-native")] #[test] - fn test_resolve_hdfs_distinct_authorities_share_instance() { + fn test_resolve_hdfs_native_distinct_authorities_share_instance() { let storage = empty_resolving_storage(); let a = storage.resolve("hdfs://ns1/a").unwrap(); @@ -426,7 +426,7 @@ mod tests { assert!( Arc::ptr_eq(&a, &b), - "different authorities should share the OpenDalStorage::Hdfs instance (operator cache is internal)" + "different authorities should share the OpenDalStorage::HdfsNative instance (operator cache is internal)" ); } diff --git a/crates/storage/opendal/tests/file_io_hdfs_test.rs b/crates/storage/opendal/tests/file_io_hdfs_test.rs index f072a9c476..d96c2eba81 100644 --- a/crates/storage/opendal/tests/file_io_hdfs_test.rs +++ b/crates/storage/opendal/tests/file_io_hdfs_test.rs @@ -33,7 +33,7 @@ mod tests { fn get_file_io() -> FileIO { set_up(); - FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Hdfs)).build() + FileIOBuilder::new(Arc::new(OpenDalStorageFactory::HdfsNative)).build() } fn test_path(suffix: &str) -> String { @@ -77,7 +77,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_configured_name_node() { set_up(); - let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Hdfs)) + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::HdfsNative)) .with_prop(HDFS_NAME_NODE, get_hdfs_endpoint()) .build(); From 9d7d2d89e8391245863ebbcbad753b99a0b3b4cc Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Sun, 13 Sep 2026 11:07:30 +0300 Subject: [PATCH 5/7] ci: retrigger after setup-python manifest timeout on macos runner Co-Authored-By: Claude Fable 5.1 From d17ae3882e82977e8f9786e287edca07310c3cf6 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Thu, 17 Sep 2026 11:36:54 +0300 Subject: [PATCH 6/7] =?UTF-8?q?fix(storage):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20cache=20newtype,=20build=20outside=20lock,=20single?= =?UTF-8?q?=20NameNode=20rule,=20env-gated=20HDFS=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - HdfsNativeOperatorCache newtype keeps the cache representation out of the public API. - Operators are built outside the cache lock (the build reads Hadoop XML synchronously); a racing first caller's duplicate is dropped unopened. - hdfs_native_effective_name_node is the single source of the configured-else-authority rule for both create_operator and the delete_stream batch key; unresolvable paths key on themselves. - hdfs.name-node is trimmed and an empty value is treated as unset, so it no longer shadows the path-authority fallback. - HDFS integration tests self-skip unless ICEBERG_TEST_HDFS_ENDPOINT is set (as the HF tests do) and the compose services sit behind the hdfs profile; the Linux CI job opts in via env. Co-Authored-By: Claude Fable 5.1 --- .github/workflows/ci.yml | 6 + crates/storage/opendal/public-api.txt | 9 +- crates/storage/opendal/src/hdfs_native.rs | 211 +++++++++++------- crates/storage/opendal/src/lib.rs | 9 +- crates/storage/opendal/src/resolving.rs | 2 +- .../opendal/tests/file_io_hdfs_test.rs | 43 +++- dev/docker-compose.yaml | 3 + 7 files changed, 197 insertions(+), 86 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f9fce8c20c..24031c1109 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -241,6 +241,9 @@ jobs: - name: Start Docker containers if: matrix.test-suite.name == 'default' + env: + # The HDFS fixture needs host networking; opt in here (Linux only). + COMPOSE_PROFILES: hdfs run: make docker-up - name: Run tests @@ -251,6 +254,9 @@ jobs: HF_TOKEN: ${{ secrets.HF_TOKEN }} HF_BUCKET: ${{ secrets.HF_BUCKET }} HF_DATASET: ${{ secrets.HF_DATASET }} + # HDFS integration tests self-skip when unset; see + # crates/storage/opendal/tests/file_io_hdfs_test.rs + ICEBERG_TEST_HDFS_ENDPOINT: hdfs://localhost:8020 run: | if [ "${{ matrix.test-suite.name }}" = "default" ]; then cargo nextest run ${{ matrix.test-suite.args }} diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index 58bd9ee6b7..b5ccaf078c 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -8,7 +8,7 @@ pub iceberg_storage_opendal::OpenDalStorage::Gcs pub iceberg_storage_opendal::OpenDalStorage::Gcs::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::HdfsNative pub iceberg_storage_opendal::OpenDalStorage::HdfsNative::config: alloc::sync::Arc -pub iceberg_storage_opendal::OpenDalStorage::HdfsNative::operators: alloc::sync::Arc>> +pub iceberg_storage_opendal::OpenDalStorage::HdfsNative::operators: iceberg_storage_opendal::HdfsNativeOperatorCache pub iceberg_storage_opendal::OpenDalStorage::Hf pub iceberg_storage_opendal::OpenDalStorage::Hf::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::LocalFs @@ -65,6 +65,13 @@ 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::HdfsNativeOperatorCache(_) +impl core::clone::Clone for iceberg_storage_opendal::HdfsNativeOperatorCache +pub fn iceberg_storage_opendal::HdfsNativeOperatorCache::clone(&self) -> iceberg_storage_opendal::HdfsNativeOperatorCache +impl core::default::Default for iceberg_storage_opendal::HdfsNativeOperatorCache +pub fn iceberg_storage_opendal::HdfsNativeOperatorCache::default() -> iceberg_storage_opendal::HdfsNativeOperatorCache +impl core::fmt::Debug for iceberg_storage_opendal::HdfsNativeOperatorCache +pub fn iceberg_storage_opendal::HdfsNativeOperatorCache::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg_storage_opendal::OpenDalResolvingStorage impl core::fmt::Debug for iceberg_storage_opendal::OpenDalResolvingStorage pub fn iceberg_storage_opendal::OpenDalResolvingStorage::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result diff --git a/crates/storage/opendal/src/hdfs_native.rs b/crates/storage/opendal/src/hdfs_native.rs index 9de9ecab5f..48d68f80ab 100644 --- a/crates/storage/opendal/src/hdfs_native.rs +++ b/crates/storage/opendal/src/hdfs_native.rs @@ -18,7 +18,7 @@ //! HDFS storage backend via OpenDAL's `services-hdfs-native` (pure Rust, no JNI). use std::collections::HashMap; -use std::sync::RwLock; +use std::sync::{Arc, RwLock}; use iceberg::io::{HDFS_HADOOP_CONF_PREFIX, HDFS_NAME_NODE}; use iceberg::{Error, ErrorKind, Result}; @@ -32,7 +32,13 @@ use crate::utils::from_opendal_error; pub(crate) fn hdfs_native_config_parse(mut m: HashMap) -> Result { let mut cfg = HdfsNativeConfig::default(); - if let Some(name_node) = m.remove(HDFS_NAME_NODE) { + // `Operator::from_config` bypasses the builder's empty-string guard, and + // `Some("")` would shadow the path-authority fallback below. + if let Some(name_node) = m + .remove(HDFS_NAME_NODE) + .map(|s| s.trim().trim_end_matches('/').to_string()) + .filter(|s| !s.is_empty()) + { cfg.name_node = Some(name_node); } @@ -86,68 +92,91 @@ pub(crate) fn hdfs_native_parse_path(path: &str) -> Result<(Option, &str Ok((name_node, rel)) } -/// Creates an operator for the path, cached per effective NameNode (the -/// configured `hdfs.name-node`, else the path authority) โ€” each operator -/// holds an HDFS client with live RPC connections. -pub(crate) fn hdfs_native_create_operator<'a>( - path: &'a str, +/// Resolves the effective NameNode for a path โ€” the configured +/// `hdfs.name-node` when set, else the path authority โ€” plus the relative +/// path. Both the operator cache and `delete_stream` batching key on this, +/// so they cannot drift apart. +pub(crate) fn hdfs_native_effective_name_node<'a>( config: &HdfsNativeConfig, - operators: &RwLock>, -) -> Result<(Operator, &'a str)> { + path: &'a str, +) -> Result<(String, &'a str)> { let (authority_name_node, relative_path) = hdfs_native_parse_path(path)?; - - let name_node = match config.name_node.clone().or(authority_name_node) { - Some(name_node) => name_node, - None => { - return Err(Error::new( + let name_node = config + .name_node + .clone() + .or(authority_name_node) + .ok_or_else(|| { + Error::new( ErrorKind::DataInvalid, format!( "Invalid hdfs path: {path}, authority-less paths require the `{HDFS_NAME_NODE}` property" ), - )); - } - }; + ) + })?; + Ok((name_node, relative_path)) +} - // Fast path: check read lock first. - { - let cache = operators - .read() - .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; - if let Some(op) = cache.get(&name_node) { - return Ok((op.clone(), relative_path)); - } +/// Operators cached per effective NameNode: each holds an `hdfs-native` +/// client with live RPC connections, whose tasks run on the tokio runtime +/// current when it was built. +#[derive(Clone, Debug, Default)] +pub struct HdfsNativeOperatorCache(Arc>>); + +impl HdfsNativeOperatorCache { + fn get(&self, name_node: &str) -> Result> { + Ok(self.0.read().map_err(poisoned)?.get(name_node).cloned()) } - // Slow path: build and insert under write lock, re-checking for a - // concurrent insert. - let mut cache = operators - .write() - .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; - let op = match cache.get(&name_node) { - Some(op) => op.clone(), - None => { - let op = hdfs_native_operator_build(config, &name_node)?; - cache.insert(name_node, op.clone()); - op - } - }; + /// Inserts `op` unless a concurrent caller got there first, returning + /// whichever operator the cache now holds. + fn insert(&self, name_node: String, op: Operator) -> Result { + Ok(self + .0 + .write() + .map_err(poisoned)? + .entry(name_node) + .or_insert(op) + .clone()) + } + + #[cfg(test)] + fn len(&self) -> usize { + self.0.read().unwrap().len() + } +} + +fn poisoned(_: T) -> Error { + Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned") +} + +/// Creates an operator for the path, reusing the cached one for its +/// effective NameNode. +pub(crate) fn hdfs_native_create_operator<'a>( + path: &'a str, + config: &HdfsNativeConfig, + operators: &HdfsNativeOperatorCache, +) -> Result<(Operator, &'a str)> { + let (name_node, relative_path) = hdfs_native_effective_name_node(config, path)?; + + if let Some(op) = operators.get(&name_node)? { + return Ok((op, relative_path)); + } - Ok((op, relative_path)) + // Built outside the lock: the build reads the Hadoop XML config + // synchronously. A racing first caller may build too; the loser is + // dropped before opening any connection. + let op = hdfs_native_operator_build(config, &name_node)?; + Ok((operators.insert(name_node, op)?, relative_path)) } /// Returns the `delete_stream` grouping key for a path: the effective -/// NameNode, mirroring the operator-cache key so paths that resolve to -/// different operators never share a deleter. +/// NameNode, so paths that resolve to different operators never share a +/// deleter. Unresolvable paths key on themselves (as `hf_batch_key` does); +/// `create_operator` then reports the real error. pub(crate) fn hdfs_native_batch_key(config: &HdfsNativeConfig, path: &str) -> String { - config - .name_node - .clone() - .or_else(|| { - hdfs_native_parse_path(path) - .ok() - .and_then(|(name_node, _)| name_node) - }) - .unwrap_or_default() + hdfs_native_effective_name_node(config, path) + .map(|(name_node, _)| name_node) + .unwrap_or_else(|_| path.to_string()) } /// Build a new OpenDAL [`Operator`]: OpenDAL splits `name_node` on commas @@ -198,6 +227,50 @@ mod tests { assert_eq!(cfg.options, None); } + #[test] + fn test_hdfs_native_config_parse_normalizes_name_node() { + let parse = |value: &str| { + hdfs_native_config_parse(HashMap::from([( + HDFS_NAME_NODE.to_string(), + value.to_string(), + )])) + .unwrap() + .name_node + }; + + // Empty must not shadow the path-authority fallback. + assert_eq!(parse(""), None); + assert_eq!(parse(" "), None); + // Trailing `/` would otherwise yield a second cache entry for one cluster. + assert_eq!( + parse(" hdfs://nn:8020/ ").as_deref(), + Some("hdfs://nn:8020") + ); + } + + #[test] + fn test_hdfs_native_effective_name_node_precedence() { + let configured = hdfs_native_config_parse(HashMap::from([( + HDFS_NAME_NODE.to_string(), + "hdfs://nn1:8020,hdfs://nn2:8020".to_string(), + )])) + .unwrap(); + let unconfigured = HdfsNativeConfig::default(); + + // Configured wins over the authority, including for authority-less paths. + for path in ["hdfs://ns-a/x", "hdfs:///y"] { + let (nn, _) = hdfs_native_effective_name_node(&configured, path).unwrap(); + assert_eq!(nn, "hdfs://nn1:8020,hdfs://nn2:8020"); + } + // Otherwise the authority, including its port. + let (nn, rel) = + hdfs_native_effective_name_node(&unconfigured, "hdfs://nn:9000/a/b").unwrap(); + assert_eq!((nn.as_str(), rel), ("hdfs://nn:9000", "a/b")); + // Neither: a pointed error. + let err = hdfs_native_effective_name_node(&unconfigured, "hdfs:///a").unwrap_err(); + assert!(err.to_string().contains(HDFS_NAME_NODE)); + } + #[test] fn test_hdfs_native_parse_path_with_authority_and_rel() { let (nn, rel) = hdfs_native_parse_path("hdfs://nameservice1/a/b").unwrap(); @@ -277,23 +350,12 @@ mod tests { } #[test] - fn test_hdfs_native_batch_key_configured_name_node_wins() { - let config = hdfs_native_config_parse(HashMap::from([( - HDFS_NAME_NODE.to_string(), - "hdfs://nn1:8020,hdfs://nn2:8020".to_string(), - )])) - .unwrap(); + fn test_hdfs_native_batch_key_invalid_path_keys_on_itself() { + let config = HdfsNativeConfig::default(); - // All paths group under the configured NameNode, matching the - // single cached operator they resolve to. - assert_eq!( - hdfs_native_batch_key(&config, "hdfs://ns-a/x"), - "hdfs://nn1:8020,hdfs://nn2:8020" - ); - assert_eq!( - hdfs_native_batch_key(&config, "hdfs:///y"), - "hdfs://nn1:8020,hdfs://nn2:8020" - ); + // Unresolvable paths must not collapse onto a shared "" key. + assert_eq!(hdfs_native_batch_key(&config, "not-a-url"), "not-a-url"); + assert_eq!(hdfs_native_batch_key(&config, "hdfs:///a"), "hdfs:///a"); } #[test] @@ -303,45 +365,44 @@ mod tests { "hdfs://configured:8020".to_string(), )])) .unwrap(); - let operators = RwLock::new(HashMap::new()); + let operators = HdfsNativeOperatorCache::default(); let (_, rel) = hdfs_native_create_operator("hdfs://from-path:9000/a/b", &config, &operators).unwrap(); assert_eq!(rel, "a/b"); - let cache = operators.read().unwrap(); - assert!(cache.contains_key("hdfs://configured:8020")); - assert!(!cache.contains_key("hdfs://from-path:9000")); + assert!(operators.get("hdfs://configured:8020").unwrap().is_some()); + assert!(operators.get("hdfs://from-path:9000").unwrap().is_none()); } #[test] fn test_hdfs_native_create_operator_uses_path_authority() { let config = HdfsNativeConfig::default(); - let operators = RwLock::new(HashMap::new()); + let operators = HdfsNativeOperatorCache::default(); let (_, rel) = hdfs_native_create_operator("hdfs://nn:8020/a/b", &config, &operators).unwrap(); assert_eq!(rel, "a/b"); - assert!(operators.read().unwrap().contains_key("hdfs://nn:8020")); + assert!(operators.get("hdfs://nn:8020").unwrap().is_some()); } #[test] fn test_hdfs_native_create_operator_caches_per_name_node() { let config = HdfsNativeConfig::default(); - let operators = RwLock::new(HashMap::new()); + let operators = HdfsNativeOperatorCache::default(); hdfs_native_create_operator("hdfs://nn1:8020/a", &config, &operators).unwrap(); hdfs_native_create_operator("hdfs://nn1:8020/b", &config, &operators).unwrap(); hdfs_native_create_operator("hdfs://nn2:8020/c", &config, &operators).unwrap(); - assert_eq!(operators.read().unwrap().len(), 2); + assert_eq!(operators.len(), 2); } #[test] fn test_hdfs_native_create_operator_authority_less_without_config_errors() { let config = HdfsNativeConfig::default(); - let operators = RwLock::new(HashMap::new()); + let operators = HdfsNativeOperatorCache::default(); let err = hdfs_native_create_operator("hdfs:///a/b", &config, &operators).unwrap_err(); diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index d3bec9d23f..19a433ca56 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -77,9 +77,8 @@ cfg_if! { cfg_if! { if #[cfg(feature = "opendal-hdfs-native")] { mod hdfs_native; - use std::sync::RwLock; - use hdfs_native::*; + pub use hdfs_native::HdfsNativeOperatorCache; use opendal::services::HdfsNativeConfig; } } @@ -197,7 +196,7 @@ impl StorageFactory for OpenDalStorageFactory { #[cfg(feature = "opendal-hdfs-native")] OpenDalStorageFactory::HdfsNative => Ok(Arc::new(OpenDalStorage::HdfsNative { config: hdfs_native_config_parse(config.props().clone())?.into(), - operators: Arc::new(RwLock::new(HashMap::new())), + operators: HdfsNativeOperatorCache::default(), })), #[cfg(feature = "opendal-oss")] OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::Oss { @@ -272,7 +271,7 @@ pub enum OpenDalStorage { config: Arc, /// Operator cache keyed by effective NameNode. #[serde(skip, default)] - operators: Arc>>, + operators: HdfsNativeOperatorCache, }, /// OSS storage variant. #[cfg(feature = "opendal-oss")] @@ -857,7 +856,7 @@ mod tests { fn hdfs_native_test_storage() -> OpenDalStorage { OpenDalStorage::HdfsNative { config: Arc::new(HdfsNativeConfig::default()), - operators: Arc::new(RwLock::new(HashMap::new())), + operators: HdfsNativeOperatorCache::default(), } } diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index b3af99928e..5e51dead48 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -134,7 +134,7 @@ fn build_storage_for_scheme( let config = crate::hdfs_native::hdfs_native_config_parse(props.clone())?; Ok(OpenDalStorage::HdfsNative { config: Arc::new(config), - operators: Arc::new(RwLock::new(HashMap::new())), + operators: crate::HdfsNativeOperatorCache::default(), }) } unsupported => Err(Error::new( diff --git a/crates/storage/opendal/tests/file_io_hdfs_test.rs b/crates/storage/opendal/tests/file_io_hdfs_test.rs index d96c2eba81..816c784189 100644 --- a/crates/storage/opendal/tests/file_io_hdfs_test.rs +++ b/crates/storage/opendal/tests/file_io_hdfs_test.rs @@ -17,9 +17,16 @@ //! Integration tests for HDFS FileIO via OpenDAL `services-hdfs-native`. //! -//! These tests need the `hdfs-namenode`/`hdfs-datanode` services from -//! `dev/docker-compose.yaml` (started by `make docker-up`); the fixture -//! uses host networking, which needs Linux or a recent Docker runtime. +//! These tests need the HDFS fixture in `dev/docker-compose.yaml` and are +//! skipped when `ICEBERG_TEST_HDFS_ENDPOINT` is not set. The fixture uses +//! host networking (Linux, or a Docker runtime that supports it), so it sits +//! behind a compose profile: +//! +//! ```text +//! COMPOSE_PROFILES=hdfs make docker-up +//! ICEBERG_TEST_HDFS_ENDPOINT=hdfs://localhost:8020 cargo test -p iceberg-storage-opendal \ +//! --features opendal-hdfs-native --test file_io_hdfs_test +//! ``` #[cfg(feature = "opendal-hdfs-native")] mod tests { @@ -29,7 +36,23 @@ mod tests { use futures::StreamExt; use iceberg::io::{FileIO, FileIOBuilder, HDFS_NAME_NODE}; use iceberg_storage_opendal::{OpenDalResolvingStorageFactory, OpenDalStorageFactory}; - use iceberg_test_utils::{get_hdfs_endpoint, normalize_test_name_with_parts, set_up}; + use iceberg_test_utils::{ + ENV_HDFS_ENDPOINT, get_hdfs_endpoint, normalize_test_name_with_parts, set_up, + }; + + /// Skips the calling test unless the HDFS fixture endpoint is configured; + /// an unset *or* empty variable means "not provided" (see the HF tests). + macro_rules! require_hdfs { + () => { + match std::env::var(ENV_HDFS_ENDPOINT) { + Ok(v) if !v.is_empty() => {} + _ => { + eprintln!("Skipping HDFS test: {} not set", ENV_HDFS_ENDPOINT); + return; + } + } + }; + } fn get_file_io() -> FileIO { set_up(); @@ -46,6 +69,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_exists() { + require_hdfs!(); let file_io = get_file_io(); let absent = test_path("test_file_io_hdfs_exists_absent"); @@ -54,6 +78,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_write_and_read() { + require_hdfs!(); let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_write_and_read"); let _ = file_io.delete(&path).await; @@ -76,6 +101,7 @@ mod tests { /// `hdfs.name-node` carries the (comma-separated) endpoints; it wins. #[tokio::test] async fn test_file_io_hdfs_configured_name_node() { + require_hdfs!(); set_up(); let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::HdfsNative)) .with_prop(HDFS_NAME_NODE, get_hdfs_endpoint()) @@ -104,6 +130,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_overwrite() { + require_hdfs!(); let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_overwrite"); let _ = file_io.delete(&path).await; @@ -125,6 +152,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_delete_stream() { + require_hdfs!(); let file_io = get_file_io(); let paths: Vec = (0..5) @@ -151,6 +179,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_delete_stream_empty() { + require_hdfs!(); let file_io = get_file_io(); let stream = futures::stream::empty().boxed(); file_io.delete_stream(stream).await.unwrap(); @@ -158,6 +187,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_resolving_storage() { + require_hdfs!(); set_up(); let file_io = FileIOBuilder::new(Arc::new(OpenDalResolvingStorageFactory::new())).build(); let path = test_path("test_file_io_hdfs_resolving_storage"); @@ -180,6 +210,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_metadata() { + require_hdfs!(); let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_metadata"); let _ = file_io.delete(&path).await; @@ -198,6 +229,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_delete() { + require_hdfs!(); let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_delete"); @@ -215,6 +247,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_delete_prefix() { + require_hdfs!(); let file_io = get_file_io(); let dir = test_path("test_file_io_hdfs_delete_prefix"); let _ = file_io.delete_prefix(&dir).await; @@ -239,6 +272,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_reader_range() { + require_hdfs!(); let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_reader_range"); let _ = file_io.delete(&path).await; @@ -264,6 +298,7 @@ mod tests { #[tokio::test] async fn test_file_io_hdfs_streaming_writer() { + require_hdfs!(); let file_io = get_file_io(); let path = test_path("test_file_io_hdfs_streaming_writer"); let _ = file_io.delete(&path).await; diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml index 15c9df875f..1356470f36 100644 --- a/dev/docker-compose.yaml +++ b/dev/docker-compose.yaml @@ -152,8 +152,10 @@ services: # ============================================================================= # hdfs-native connects to DataNodes by their registered IP โ€” unroutable on # a docker bridge, hence host networking (needs Linux or a recent runtime). + # Opt-in: `COMPOSE_PROFILES=hdfs make docker-up`. hdfs-namenode: image: apache/hadoop:3.5.0 + profiles: [hdfs] network_mode: "host" command: ["hdfs", "namenode"] environment: @@ -172,6 +174,7 @@ services: hdfs-datanode: image: apache/hadoop:3.5.0 + profiles: [hdfs] network_mode: "host" command: ["hdfs", "datanode"] depends_on: From b8cc860d638a05bdb5f30439a0d1450a1902b0d9 Mon Sep 17 00:00:00 2001 From: Michael Taranov Date: Thu, 17 Sep 2026 11:36:54 +0300 Subject: [PATCH 7/7] chore(deps): bump rustls to 0.23.45 (RUSTSEC-2026-0285) Inherited from main (tracked in #3222); same bump as #3165. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 237cbb6889..1c117dc921 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -638,9 +638,9 @@ dependencies = [ [[package]] name = "aws-lc-rs" -version = "1.17.1" +version = "1.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4342d8937fc7e5dd9b1c60292261c0670c882a2cd1719cfc11b1af41731e32ad" +checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" dependencies = [ "aws-lc-sys", "zeroize", @@ -648,9 +648,9 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.42.0" +version = "0.45.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d9ceb1da931507a12f4fccea479dccd00da1943e1b4ae72d8e502d707361444" +checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" dependencies = [ "cc", "cmake", @@ -6769,9 +6769,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.41" +version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ "aws-lc-rs", "once_cell", @@ -6833,9 +6833,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.13" +version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ "aws-lc-rs", "ring", @@ -7818,7 +7818,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2",