diff --git a/Cargo.lock b/Cargo.lock index 3a5c79a550..55b1eb463d 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..97673c15fb 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1015,6 +1015,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..e2d468dec1 --- /dev/null +++ b/crates/iceberg/src/io/storage/config/hdfs.rs @@ -0,0 +1,80 @@ +// 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 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. +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. +// 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/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..9abfa70251 --- /dev/null +++ b/crates/storage/opendal/src/hdfs.rs @@ -0,0 +1,349 @@ +// 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}"), + ) + })?; + // 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() + .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 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)) +} + +/// 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 { + 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_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([( + 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..9e01793077 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( @@ -377,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())) @@ -434,6 +474,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 +519,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 +796,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..f072a9c476 --- /dev/null +++ b/crates/storage/opendal/tests/file_io_hdfs_test.rs @@ -0,0 +1,280 @@ +// 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` (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 { + 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] + 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] + 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] + 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] + 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] + 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] + 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] + 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] + 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] + 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] + 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] + 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] + 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..485adb900a 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 (needs Linux or a recent runtime). + 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 + +