diff --git a/Cargo.lock b/Cargo.lock index 75694c3875..9f35da262d 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", ] @@ -1135,6 +1146,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" @@ -1268,7 +1288,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]] @@ -1343,7 +1372,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]] @@ -1523,6 +1563,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" @@ -1687,7 +1733,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]] @@ -1864,6 +1919,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" @@ -1941,6 +2005,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" @@ -2267,6 +2343,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" @@ -2443,6 +2547,47 @@ dependencies = [ "hashbrown 0.15.5", ] +[[package]] +name = "hdfs-native" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd181084003308224efddf737417832186839ce2882d2468ddad0114cfb3551" +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" @@ -2958,6 +3103,7 @@ dependencies = [ "serde_json", "tempfile", "tokio", + "tracing", "typetag", "url", ] @@ -3109,10 +3255,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 = "4.1.0" @@ -3371,6 +3527,16 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libloading" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "754ca22de805bb5744484a5b151a9e1a8e837d5dc232c2d7d8c2e3492edc8b60" +dependencies = [ + "cfg-if 1.0.4", + "windows-link", +] + [[package]] name = "libm" version = "0.2.16" @@ -3941,6 +4107,7 @@ dependencies = [ "opendal-service-azdls", "opendal-service-fs", "opendal-service-gcs", + "opendal-service-hdfs-native", "opendal-service-hf", "opendal-service-oss", "opendal-service-s3", @@ -4095,6 +4262,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" @@ -4423,8 +4604,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", @@ -4584,6 +4765,38 @@ dependencies = [ "yansi", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools 0.13.0", + "proc-macro2", + "quote", + "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 = "ptr_meta" version = "0.3.1" @@ -5234,6 +5447,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" @@ -5425,7 +5647,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" dependencies = [ - "cipher", + "cipher 0.4.4", ] [[package]] @@ -7288,6 +7510,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" @@ -7321,13 +7552,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" @@ -7349,6 +7597,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" @@ -7361,6 +7615,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" @@ -7373,12 +7633,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" @@ -7391,6 +7663,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" @@ -7403,6 +7681,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" @@ -7415,6 +7699,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" @@ -7427,6 +7717,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 = "1.0.3" diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 8a34bbf7b5..011250e043 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1072,6 +1072,10 @@ pub const iceberg::io::GCS_PROJECT_ID: &str pub const iceberg::io::GCS_SERVICE_HOST: &str pub const iceberg::io::GCS_TOKEN: &str pub const iceberg::io::GCS_USER_PROJECT: &str +pub const iceberg::io::HDFS_HADOOP_CONF_PREFIX: &str +pub const iceberg::io::HDFS_HOST: &str +pub const iceberg::io::HDFS_NAME_NODE: &str +pub const iceberg::io::HDFS_PORT: &str pub const iceberg::io::HF_ENDPOINT: &str pub const iceberg::io::HF_REVISION: &str pub const iceberg::io::HF_TOKEN: &str diff --git a/crates/iceberg/src/io/storage/config/hdfs_native.rs b/crates/iceberg/src/io/storage/config/hdfs_native.rs new file mode 100644 index 0000000000..ef9e16bf65 --- /dev/null +++ b/crates/iceberg/src/io/storage/config/hdfs_native.rs @@ -0,0 +1,45 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! HDFS storage configuration. + +/// HDFS NameNode RPC endpoint(s) as `host:port`, e.g. `hdfs://namenode:8020`; +/// a comma-separated list enables HA failover. Serves authority-less paths +/// (`hdfs:///path`) only; a path authority with a port is used as is. A +/// portless authority (`hdfs://ns1/path`) must be a nameservice declared in +/// properties: `hdfs.name-node.` takes the same list and is +/// sugar for Hadoop's `dfs.ha.namenodes.` and +/// `dfs.namenode.rpc-address..`, which are honored as well +/// when passed through [`HDFS_HADOOP_CONF_PREFIX`]. Unlike Hadoop, an +/// undeclared one is an error: it is neither resolved from +/// `$HADOOP_CONF_DIR` nor dialed on the default port 8020. +pub const HDFS_NAME_NODE: &str = "hdfs.name-node"; +/// NameNode host for authority-less paths, as in PyIceberg; paths that carry +/// an authority ignore it, as they do there. Combined with [`HDFS_PORT`] into +/// Hadoop's `fs.defaultFS`. PyIceberg's `hdfs.user` and `hdfs.kerberos_ticket` +/// have no equivalent: the client reads `HADOOP_USER_NAME` and the default +/// Kerberos credential cache. +pub const HDFS_HOST: &str = "hdfs.host"; +/// NameNode port for [`HDFS_HOST`]; defaults to `8020` and has no effect +/// without the host. +pub const HDFS_PORT: &str = "hdfs.port"; +/// Prefix for properties forwarded to the HDFS client configuration with the +/// prefix stripped: `hadoop.dfs.client.failover.random.order` is forwarded as +/// `dfs.client.failover.random.order`. Forwarded values override those loaded +/// from `$HADOOP_CONF_DIR`; `hadoop.fs.defaultFS` also serves authority-less +/// paths. +pub const HDFS_HADOOP_CONF_PREFIX: &str = "hadoop."; diff --git a/crates/iceberg/src/io/storage/config/mod.rs b/crates/iceberg/src/io/storage/config/mod.rs index d8d356de16..8ea4af080e 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_native; mod hf; mod oss; mod s3; @@ -40,6 +41,7 @@ use std::collections::HashMap; pub use azdls::*; pub use gcs::*; +pub use hdfs_native::*; pub use hf::*; pub use oss::*; pub use s3::*; diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index d3403598e8..83541e89c5 100644 --- a/crates/storage/opendal/Cargo.toml +++ b/crates/storage/opendal/Cargo.toml @@ -42,6 +42,11 @@ opendal-all = [ opendal-azdls = ["opendal/services-azdls"] opendal-fs = ["opendal/services-fs"] opendal-gcs = ["opendal/services-gcs"] +opendal-hdfs-native = [ + "opendal/services-hdfs-native", + "dep:tokio", + "dep:tracing", +] opendal-hf = ["opendal/services-hf"] opendal-memory = ["opendal/services-memory"] opendal-oss = ["opendal/services-oss"] @@ -59,6 +64,8 @@ opendal = { workspace = true } reqsign-aws-v4 = { version = "3.0.0", optional = true } reqsign-core = { version = "3.0.0", optional = true } serde = { workspace = true } +tokio = { workspace = true, optional = true } +tracing = { workspace = true, optional = true } typetag = { workspace = true } url = { workspace = true } diff --git a/crates/storage/opendal/README.md b/crates/storage/opendal/README.md index 5db4293e67..a254433f92 100644 --- a/crates/storage/opendal/README.md +++ b/crates/storage/opendal/README.md @@ -32,11 +32,14 @@ OpenDAL-based storage backend implementations for [Apache Iceberg Rust](https:// | Hugging Face | `opendal-hf` | ✅ Stable | Hugging Face buckets and repositories | | Alibaba Cloud OSS | `opendal-oss` | 🧪 Experimental | Alibaba Cloud Object Storage Service | | Azure Datalake | `opendal-azdls` | 🧪 Experimental | Azure Datalake Storage v2 | +| HDFS | `opendal-hdfs-native` | 🧪 Experimental | HDFS via the native Rust client (`hdfs-native`) | You can enable all stable storage backends at once using the `opendal-all` feature flag. > Note that `opendal-oss` and `opendal-azdls` are currently experimental and not included in `opendal-all`. +> `opendal-hdfs-native` is experimental and not included in `opendal-all`; see [HDFS](#hdfs). + ## Usage Add the crate to your `Cargo.toml` with the feature flags for the backends you need: @@ -82,3 +85,40 @@ async fn main() -> iceberg::Result<()> { Ok(()) } ``` + +## HDFS + +The `opendal-hdfs-native` feature adds an HDFS backend built on +[`hdfs-native`](https://github.com/Kimahriman/hdfs-native), a native Rust client that needs no +JVM. Enable it on its own, since it is not part of `opendal-all`: + +```toml +iceberg-storage-opendal = { version = "x.y.z", features = ["opendal-hdfs-native"] } +``` + +Then pass `OpenDalStorageFactory::HdfsNative` to your catalog builder, or use +`OpenDalResolvingStorageFactory`, which routes `hdfs://` paths to it. The backend reads these +properties: + +| Property | Description | +| --- | --- | +| `hdfs.name-node` | NameNode `host:port` for authority-less paths (`hdfs:///warehouse/...`); a comma-separated list enables HA failover | +| `hdfs.name-node.` | The NameNodes of a nameservice used as a path authority (`hdfs://nameservice1/warehouse/...`), comma-separated | +| `hdfs.host`, `hdfs.port` | PyIceberg's keys for authority-less paths; the port defaults to `8020` | +| `hadoop.*` | Forwarded to the HDFS client with the prefix stripped, e.g. `hadoop.dfs.client.failover.max.attempts` | + +A path with a `host:port` authority is always used as is. For a table under +`hdfs://nameservice1/warehouse` on an HA cluster: + +```text +hdfs.name-node.nameservice1 = nn1.example.com:8020,nn2.example.com:8020 +``` + +The client also loads `core-site.xml` and `hdfs-site.xml` from `$HADOOP_CONF_DIR`, else +`$HADOOP_HOME/etc/hadoop`, and takes the user from `HADOOP_USER_NAME`. Kerberos is used when +`hadoop.security.authentication` is `kerberos` there, with the default credential cache +(`kinit`); `libgssapi_krb5` is loaded at runtime. PyIceberg's `hdfs.user` and +`hdfs.kerberos_ticket` are ignored with a warning. + +Unlike Hadoop, a portless authority is never resolved from `$HADOOP_CONF_DIR` or dialed on +port 8020: declare it with `hdfs.name-node.`. diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index c2e310335c..797d90d1e8 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -8,6 +8,7 @@ pub iceberg_storage_opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc +pub iceberg_storage_opendal::OpenDalStorage::HdfsNative(iceberg_storage_opendal::HdfsNativeStorage) pub iceberg_storage_opendal::OpenDalStorage::Hf pub iceberg_storage_opendal::OpenDalStorage::Hf::client_config: iceberg_storage_opendal::OpenDalClientConfig pub iceberg_storage_opendal::OpenDalStorage::Hf::config: alloc::sync::Arc @@ -47,6 +48,7 @@ pub enum iceberg_storage_opendal::OpenDalStorageFactory pub iceberg_storage_opendal::OpenDalStorageFactory::Azdls pub iceberg_storage_opendal::OpenDalStorageFactory::Fs pub iceberg_storage_opendal::OpenDalStorageFactory::Gcs +pub iceberg_storage_opendal::OpenDalStorageFactory::HdfsNative pub iceberg_storage_opendal::OpenDalStorageFactory::Hf pub iceberg_storage_opendal::OpenDalStorageFactory::Memory pub iceberg_storage_opendal::OpenDalStorageFactory::Oss @@ -69,6 +71,15 @@ impl core::clone::Clone for iceberg_storage_opendal::CustomAwsCredentialLoader pub fn iceberg_storage_opendal::CustomAwsCredentialLoader::clone(&self) -> Self impl core::fmt::Debug for iceberg_storage_opendal::CustomAwsCredentialLoader pub fn iceberg_storage_opendal::CustomAwsCredentialLoader::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub struct iceberg_storage_opendal::HdfsNativeStorage +impl core::clone::Clone for iceberg_storage_opendal::HdfsNativeStorage +pub fn iceberg_storage_opendal::HdfsNativeStorage::clone(&self) -> iceberg_storage_opendal::HdfsNativeStorage +impl core::fmt::Debug for iceberg_storage_opendal::HdfsNativeStorage +pub fn iceberg_storage_opendal::HdfsNativeStorage::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg_storage_opendal::HdfsNativeStorage +pub fn iceberg_storage_opendal::HdfsNativeStorage::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_opendal::HdfsNativeStorage +pub fn iceberg_storage_opendal::HdfsNativeStorage::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_storage_opendal::OpenDalClientConfig impl iceberg_storage_opendal::OpenDalClientConfig pub fn iceberg_storage_opendal::OpenDalClientConfig::from_properties(properties: &std::collections::hash::map::HashMap) -> iceberg::error::Result diff --git a/crates/storage/opendal/src/hdfs_native.rs b/crates/storage/opendal/src/hdfs_native.rs new file mode 100644 index 0000000000..7e7b4f9cc9 --- /dev/null +++ b/crates/storage/opendal/src/hdfs_native.rs @@ -0,0 +1,1356 @@ +// 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::{Arc, LazyLock, PoisonError, RwLock}; + +use iceberg::io::{HDFS_HADOOP_CONF_PREFIX, HDFS_HOST, HDFS_NAME_NODE, HDFS_PORT}; +use iceberg::{Error, ErrorKind, Result}; +use iceberg_property_macro::Properties; +use opendal::Operator; +use opendal::services::HdfsNativeConfig; +use serde::{Deserialize, Serialize}; +use tokio::runtime::Runtime; +use url::Url; + +use crate::OpenDalClientConfig; +use crate::utils::from_opendal_error; + +/// Hadoop's default filesystem, which serves authority-less paths. +const FS_DEFAULT_FS: &str = "fs.defaultFS"; +const HDFS_DEFAULT_PORT: u16 = 8020; +/// PyIceberg keys with no equivalent in opendal's config. +const HDFS_UNSUPPORTED_KEYS: [&str; 2] = ["hdfs.user", "hdfs.kerberos_ticket"]; +/// Hadoop's keys declaring an HA nameservice: `hdfs.name-node.` +/// expands to them, and they are honored as well when passed through `hadoop.`. +const HA_NAMENODES_PREFIX: &str = "dfs.ha.namenodes"; +const HA_NAMENODE_RPC_ADDRESS_PREFIX: &str = "dfs.namenode.rpc-address"; + +/// Normalizes one NameNode spelling to `hdfs://host:port`, the form path +/// authorities take, so every source shares cache keys. `hdfs-native` dials +/// a socket address and has no default port, so anything else is `None`: +/// another scheme, a logical name, port 0, userinfo, a path. +fn hdfs_native_name_node(entry: &str) -> Option { + let rest = entry.trim().trim_end_matches('/'); + let rest = rest.strip_prefix("hdfs://").unwrap_or(rest); + if rest.is_empty() || rest.contains("://") { + return None; + } + let url = Url::parse(&format!("hdfs://{rest}")).ok()?; + let (host, port) = (url.host_str()?, url.port()?); + let plain = host.is_empty() + || port == 0 + || !url.username().is_empty() + || url.password().is_some() + || !matches!(url.path(), "" | "/") + || url.query().is_some() + || url.fragment().is_some(); + (!plain).then(|| hdfs_native_endpoint(host, port)) +} + +/// An endpoint as every source spells it, the host lowercased because host +/// names are case-insensitive: `NN1:8020` and `nn1:8020` share one client, +/// as they share one `FileSystem` in Hadoop's cache. +fn hdfs_native_endpoint(host: &str, port: u16) -> String { + format!("hdfs://{}:{port}", host.to_ascii_lowercase()) +} + +/// Parses a comma-separated NameNode list, each entry normalized; an empty +/// list is `Ok` and empty. +fn hdfs_native_name_node_list(property: &str, value: &str) -> Result> { + value + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(|entry| { + hdfs_native_name_node(entry).ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid `{property}` entry: {}, expected host:port (hdfs:// optional)", + hdfs_native_redact(entry) + ), + ) + }) + }) + .collect() +} + +/// Prefix of `hdfs.name-node.`, which declares one HA nameservice. +const HDFS_NAME_NODE_DECLARATION_PREFIX: &str = "hdfs.name-node."; + +/// The properties this backend reads; unknown keys are ignored. Blank values +/// count as unset, as in PyIceberg. +#[derive(Properties)] +struct HdfsNativeProperties { + /// `hdfs.name-node`, each entry normalized. + #[property( + key = HDFS_NAME_NODE, + default = Vec::new(), + parse_with = parse_name_node_list + )] + name_node: Vec, + /// `hdfs.name-node.` declarations, keyed by nameservice. + #[property(prefix = HDFS_NAME_NODE_DECLARATION_PREFIX)] + nameservices: HashMap, + #[property(key = HDFS_HOST, default = "", parse_with = parse_trimmed)] + host: String, + #[property(key = HDFS_PORT, default = "", parse_with = parse_trimmed)] + port: String, + /// `hadoop.*` keys with the prefix stripped, forwarded to the client. + #[property(prefix = HDFS_HADOOP_CONF_PREFIX)] + hadoop: HashMap, +} + +// Entries are trimmed one by one: opendal splits the list on `,` as is, so a +// space after a comma would break failover to that NameNode. An empty list +// stays unset because `Operator::from_config` bypasses the builder's +// empty-string guard and `Some("")` would shadow the path-authority fallback. +fn parse_name_node_list(value: &str) -> Result> { + hdfs_native_name_node_list(HDFS_NAME_NODE, value) +} + +fn parse_trimmed(value: &str) -> Result { + Ok(value.trim().to_string()) +} + +/// Parse iceberg properties to [`HdfsNativeConfig`]; dropped properties are +/// logged as warnings. +pub(crate) fn hdfs_native_config_parse(m: HashMap) -> Result { + hdfs_native_config_parse_with(m, |warning| tracing::warn!("{warning}")) +} + +/// The parser proper, with the warning sink injected so tests can see what +/// was dropped without a tracing subscriber. +fn hdfs_native_config_parse_with( + m: HashMap, + mut warn: impl FnMut(String), +) -> Result { + let props = HdfsNativeProperties::from_properties(&m)?; + let mut options = props.hadoop; + + // `hdfs.name-node.` is sugar for Hadoop's own declaration of + // an HA nameservice, which the resolver reads back from the options. + // Explicit `hadoop.` keys win over the sugar. + for (nameservice, value) in props.nameservices { + let key = format!("{HDFS_NAME_NODE_DECLARATION_PREFIX}{nameservice}"); + if nameservice.is_empty() || nameservice.chars().any(char::is_whitespace) { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid property `{key}`: a nameservice name must follow `{HDFS_NAME_NODE_DECLARATION_PREFIX}`" + ), + )); + } + let entries = hdfs_native_name_node_list(&key, &value)?; + if entries.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid property `{key}`: no NameNodes"), + )); + } + let ids: Vec = (0..entries.len()).map(|i| format!("nn{i}")).collect(); + options + .entry(format!("{HA_NAMENODES_PREFIX}.{nameservice}")) + .or_insert_with(|| ids.join(",")); + for (id, entry) in ids.iter().zip(&entries) { + options + .entry(format!( + "{HA_NAMENODE_RPC_ADDRESS_PREFIX}.{nameservice}.{id}" + )) + .or_insert_with(|| entry.trim_start_matches("hdfs://").to_string()); + } + } + + // A config carried over from PyIceberg would otherwise change identity + // silently; the client reads `HADOOP_USER_NAME` and the Kerberos cache. + for key in HDFS_UNSUPPORTED_KEYS { + if m.contains_key(key) { + warn(format!( + "`{key}` is not supported by the hdfs-native backend and is ignored" + )); + } + } + if m.contains_key(HDFS_HADOOP_CONF_PREFIX) { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid property `{HDFS_HADOOP_CONF_PREFIX}`: a Hadoop key must follow the prefix" + ), + )); + } + + let port = match props.port.as_str() { + "" => None, + port => Some(port.parse::().map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid `{HDFS_PORT}`: {port}"), + ) + .with_source(e) + })?), + }; + // PyIceberg's `hdfs.host`/`hdfs.port` name the filesystem for + // authority-less paths, which is what Hadoop's `fs.defaultFS` means; an + // explicit `hadoop.fs.defaultFS` wins. + match props.host.as_str() { + "" if port.is_some() => { + warn(format!( + "`{HDFS_PORT}` has no effect without `{HDFS_HOST}` and is ignored" + )); + } + "" => {} + host => { + // An IPv6 literal needs brackets in a URI authority. + let bracketed = if host.contains(':') && !host.starts_with('[') { + format!("[{host}]") + } else { + host.to_string() + }; + let port = port.unwrap_or(HDFS_DEFAULT_PORT); + // Validated like every other NameNode spelling, so a host that + // carries a scheme or port fails here and not at the first I/O. + let default_fs = hdfs_native_name_node(&format!("hdfs://{bracketed}:{port}")) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid `{HDFS_HOST}`/`{HDFS_PORT}`: {}, expected a host name or IP and a port", + hdfs_native_redact(&format!("{host}:{port}")) + ), + ) + })?; + options + .entry(FS_DEFAULT_FS.to_string()) + .or_insert(default_fs); + } + } + + let mut cfg = HdfsNativeConfig::default(); + if !props.name_node.is_empty() { + cfg.name_node = Some(props.name_node.join(",")); + } + if !options.is_empty() { + cfg.options = Some(options); + } + + // Every declaration, sugar or passed through, and the default filesystem + // fail here rather than at the first path that uses them. + for nameservice in cfg.options.iter().flat_map(|options| { + options.keys().filter_map(|key| { + key.strip_prefix(HA_NAMENODES_PREFIX) + .and_then(|rest| rest.strip_prefix('.')) + }) + }) { + hdfs_native_nameservice(&cfg, nameservice)?; + } + hdfs_native_default_name_node(&cfg)?; + Ok(cfg) +} + +/// `s` with everything between its scheme and its last `@` masked, so an +/// error message never echoes userinfo, even a password holding a raw `/`, +/// `?`, `#` or `://` that a URL parser would split elsewhere. An `@` further +/// along the path is masked the same way, which only costs context. +fn hdfs_native_redact(s: &str) -> String { + // Only a real scheme is kept; anything else before `://` may be userinfo. + let (scheme, rest) = match s.split_once("://") { + Some((scheme, rest)) + if scheme + .chars() + .all(|c| c.is_ascii_alphanumeric() || "+-.".contains(c)) => + { + (format!("{scheme}://"), rest) + } + _ => (String::new(), s), + }; + match rest.rsplit_once('@') { + Some((_, after)) => format!("{scheme}***@{after}"), + None => s.to_string(), + } +} + +/// A `DataInvalid` error for an HDFS path, its userinfo masked. +fn hdfs_native_invalid_path(path: &str, reason: impl std::fmt::Display) -> Error { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {}, {reason}", hdfs_native_redact(path)), + ) +} + +/// A path's authority, as the resolver reads it. +#[derive(Debug, PartialEq)] +pub(crate) enum HdfsNativeAuthority { + /// `hdfs://host:port`, normalized: dialed as is. + NameNode(String), + /// `hdfs://name` without a port: a nameservice that must be declared. + Nameservice(String), + /// `hdfs:///path`: served by `hdfs.name-node`, else `fs.defaultFS`. + Default, +} + +/// Parse an HDFS path into its authority and the relative path (no leading +/// `/`, opendal style). +pub(crate) fn hdfs_native_parse_path(path: &str) -> Result<(HdfsNativeAuthority, &str)> { + let url = Url::parse(path).map_err(|e| hdfs_native_invalid_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(hdfs_native_invalid_path(path, "expected scheme `hdfs://`")); + }; + // Userinfo has nowhere to go and port 0 cannot be dialed; silently + // dropping the one or reading the other as a logical name would mislead. + if !url.username().is_empty() || url.password().is_some() { + return Err(hdfs_native_invalid_path(path, "userinfo is not supported")); + } + if url.port() == Some(0) { + return Err(hdfs_native_invalid_path(path, "port 0 cannot be dialed")); + } + + let authority = match (url.host_str().filter(|h| !h.is_empty()), url.port()) { + (Some(host), Some(port)) => HdfsNativeAuthority::NameNode(hdfs_native_endpoint(host, port)), + (Some(host), None) => HdfsNativeAuthority::Nameservice(host.to_string()), + (None, _) => HdfsNativeAuthority::Default, + }; + + // `url.path()` borrows from `url` and can't be returned with the input's + // lifetime. Take the path from the original input instead: everything + // after the first `/` following the `hdfs://` prefix. Opendal paths must + // not start with `/` (`Deleter::delete` rejects them). + let rel = after_scheme + .split_once('/') + .map_or("", |(_, path)| path.trim_start_matches('/')); + + Ok((authority, rel)) +} + +/// Resolves the effective NameNode for a path, plus the relative path. An +/// authority with a port is used as is, a portless one must be a declared +/// nameservice, and an authority-less path uses `hdfs.name-node`, else +/// `fs.defaultFS`. The operator cache, `delete_stream` batching and +/// `relativize_path` all go through this, so they cannot drift apart. +pub(crate) fn hdfs_native_effective_name_node<'a>( + config: &HdfsNativeConfig, + path: &'a str, +) -> Result<(String, &'a str)> { + let (authority, relative_path) = hdfs_native_parse_path(path)?; + let name_node = match authority { + HdfsNativeAuthority::NameNode(name_node) => name_node, + HdfsNativeAuthority::Nameservice(nameservice) => { + hdfs_native_nameservice(config, &nameservice)?.ok_or_else(|| { + hdfs_native_invalid_path( + path, + format!( + "`{nameservice}` has no port and is not a declared nameservice; add the port or set `{HDFS_NAME_NODE}.{nameservice}`" + ), + ) + })? + } + HdfsNativeAuthority::Default => hdfs_native_default_name_node(config)?.ok_or_else(|| { + hdfs_native_invalid_path( + path, + format!("authority-less paths require `{HDFS_NAME_NODE}` or `{HDFS_HOST}`"), + ) + })?, + }; + Ok((name_node, relative_path)) +} + +/// The NameNode that serves authority-less paths: `hdfs.name-node`, else +/// `fs.defaultFS`, which must then be an HDFS `host:port`. +fn hdfs_native_default_name_node(config: &HdfsNativeConfig) -> Result> { + if let Some(name_node) = &config.name_node { + return Ok(Some(name_node.clone())); + } + let Some(default_fs) = config + .options + .as_ref() + .and_then(|options| options.get(FS_DEFAULT_FS)) + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + else { + return Ok(None); + }; + hdfs_native_name_node(default_fs).map(Some).ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid `{FS_DEFAULT_FS}`: {} is not an HDFS host:port, a logical nameservice requires `{HDFS_NAME_NODE}`", + hdfs_native_redact(default_fs) + ), + ) + }) +} + +/// The NameNodes that Hadoop's keys in the forwarded options declare for a +/// nameservice, if any. A declaration with a missing or malformed address is +/// an error rather than a silent fallback. +fn hdfs_native_nameservice(config: &HdfsNativeConfig, nameservice: &str) -> Result> { + let Some(options) = config.options.as_ref() else { + return Ok(None); + }; + let Some(ids) = options.get(&format!("{HA_NAMENODES_PREFIX}.{nameservice}")) else { + return Ok(None); + }; + let mut name_nodes = Vec::new(); + for id in ids.split(',').map(str::trim).filter(|id| !id.is_empty()) { + let key = format!("{HA_NAMENODE_RPC_ADDRESS_PREFIX}.{nameservice}.{id}"); + let Some(name_node) = options + .get(&key) + .and_then(|value| hdfs_native_name_node(value)) + else { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Nameservice `{nameservice}` declares NameNode `{id}` but `{key}` is missing or not host:port" + ), + )); + }; + name_nodes.push(name_node); + } + if name_nodes.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Nameservice `{nameservice}` declares no NameNodes"), + )); + } + Ok(Some(name_nodes.join(","))) +} + +/// State of [`OpenDalStorage::HdfsNative`](crate::OpenDalStorage::HdfsNative): +/// the parsed configuration and the per-NameNode operator cache. Only the +/// storage factories build it. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct HdfsNativeStorage { + pub(crate) config: Arc, + #[serde(skip, default)] + pub(crate) operators: HdfsNativeOperatorCache, + #[serde(default)] + pub(crate) client_config: OpenDalClientConfig, +} + +impl HdfsNativeStorage { + pub(crate) fn new(config: HdfsNativeConfig, client_config: OpenDalClientConfig) -> Self { + Self { + config: Arc::new(config), + operators: HdfsNativeOperatorCache::default(), + client_config, + } + } +} + +/// Operators cached per effective NameNode: one per endpoint or nameservice +/// the storage's paths reach, which in practice is a handful of clusters, so +/// it is not bounded. The cache lives as long as the storage that owns it +/// (clones share it). +#[derive(Clone, Debug, Default)] +pub(crate) struct HdfsNativeOperatorCache(Arc>>); + +// A poisoned lock is recovered rather than reported: writes only insert +// whole entries, so a panic elsewhere cannot leave the map broken. +impl HdfsNativeOperatorCache { + pub(crate) fn get(&self, name_node: &str) -> Option { + self.0 + .read() + .unwrap_or_else(PoisonError::into_inner) + .get(name_node) + .cloned() + } + + /// Inserts `op` unless a concurrent caller got there first, returning + /// whichever operator the cache now holds. + fn insert(&self, name_node: String, op: Operator) -> Operator { + self.0 + .write() + .unwrap_or_else(PoisonError::into_inner) + .entry(name_node) + .or_insert(op) + .clone() + } + + #[cfg(test)] + fn len(&self) -> usize { + self.0.read().unwrap().len() + } +} + +/// The runtime `hdfs-native` clients are built in. A client spawns its +/// NameNode connections on the runtime current when it is built and panics +/// once that runtime is gone, so building it here keeps a cached operator +/// usable from any caller's runtime. Started on first use. +// TODO: pass a runtime to the client instead of entering this one once +// opendal's hdfs-native service exposes hdfs-native's +// `ClientBuilder::with_io_runtime` (Kimahriman/hdfs-native#255). +static HDFS_NATIVE_RUNTIME: LazyLock = LazyLock::new(|| { + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .thread_name("iceberg-hdfs-native") + .build() + .expect("failed to start the HDFS client runtime") +}); + +/// 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)); + } + + // Built outside the lock, so a racing first caller may build too; the + // loser is dropped before opening any connection. The build reads the + // Hadoop XML config synchronously (~0.1 ms locally), once per NameNode + // per storage, which is accepted here rather than offloaded. + let op = { + let _runtime = HDFS_NATIVE_RUNTIME.enter(); + 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, 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 { + 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 +/// into a synthetic HA name service; `$HADOOP_CONF_DIR` XML still merges in. +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) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_hdfs_native_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_native_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_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_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); + assert_eq!(parse(" , "), None); + // `hdfs://` is optional (Hadoop's own rpc-address format is bare) and + // every entry is normalized to it, so it keys the cache like a path + // authority does; a space after a comma would otherwise break + // failover to the entry behind it. + for (value, normalized) in [ + (" hdfs://nn:8020/ ", "hdfs://nn:8020"), + ("nn:8020", "hdfs://nn:8020"), + ("[::1]:8020", "hdfs://[::1]:8020"), + ("NN:8020", "hdfs://nn:8020"), + ( + "hdfs://nn1:8020/, nn2:8020,", + "hdfs://nn1:8020,hdfs://nn2:8020", + ), + ] { + assert_eq!(parse(value).as_deref(), Some(normalized), "{value}"); + } + } + + #[test] + fn test_hdfs_native_config_parse_host_port_as_default_fs() { + let parse = |props: &[(&str, &str)]| { + let props = props + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + hdfs_native_config_parse(props) + .map(|cfg| cfg.options.and_then(|o| o.get(FS_DEFAULT_FS).cloned())) + }; + + assert_eq!( + parse(&[(HDFS_HOST, "nn"), (HDFS_PORT, " 9000 ")]) + .unwrap() + .as_deref(), + Some("hdfs://nn:9000") + ); + assert_eq!( + parse(&[(HDFS_HOST, " nn ")]).unwrap().as_deref(), + Some("hdfs://nn:8020") + ); + assert_eq!( + parse(&[(HDFS_HOST, "::1")]).unwrap().as_deref(), + Some("hdfs://[::1]:8020") + ); + // An explicit `hadoop.fs.defaultFS` wins. + assert_eq!( + parse(&[ + (HDFS_HOST, "nn"), + ("hadoop.fs.defaultFS", "hdfs://explicit:8020") + ]) + .unwrap() + .as_deref(), + Some("hdfs://explicit:8020") + ); + // An empty host is unset; a port without a host has nothing to apply to. + assert_eq!( + parse(&[(HDFS_HOST, " "), (HDFS_PORT, "9000")]).unwrap(), + None + ); + assert_eq!(parse(&[(HDFS_PORT, "9000")]).unwrap(), None); + // A bad port is rejected whether or not a host accompanies it. + for props in [ + &[(HDFS_HOST, "nn"), (HDFS_PORT, "x")][..], + &[(HDFS_PORT, "x")][..], + &[(HDFS_PORT, "70000")][..], + ] { + let err = parse(props).unwrap_err(); + assert!(err.to_string().contains(HDFS_PORT), "{props:?}: {err}"); + // The parse error rides along as the source, not in the message. + assert!( + std::error::Error::source(&err).is_some(), + "{props:?}: {err}" + ); + } + // The composed value is validated like any other NameNode spelling. + for props in [ + &[(HDFS_HOST, "nn:8020")][..], + &[(HDFS_HOST, "hdfs://nn")][..], + &[(HDFS_HOST, "nn"), (HDFS_PORT, "0")][..], + &[(HDFS_HOST, "nn/path")][..], + ] { + let err = parse(props).unwrap_err(); + assert!(err.to_string().contains(HDFS_HOST), "{props:?}: {err}"); + } + } + + #[test] + fn test_hdfs_native_config_parse_rejects_bare_prefix() { + let err = hdfs_native_config_parse(HashMap::from([( + HDFS_HADOOP_CONF_PREFIX.to_string(), + "x".to_string(), + )])) + .unwrap_err(); + assert!(err.to_string().contains(HDFS_HADOOP_CONF_PREFIX), "{err}"); + } + + /// PyIceberg's identity keys and a port without a host are dropped, but + /// visibly: each leaves a warning naming the key, and none is forwarded. + #[test] + fn test_hdfs_native_config_parse_warns_about_ignored_keys() { + let mut warnings = Vec::new(); + let cfg = hdfs_native_config_parse_with( + HashMap::from([ + ("hdfs.user".to_string(), "alice".to_string()), + ( + "hdfs.kerberos_ticket".to_string(), + "/tmp/krb5cc".to_string(), + ), + (HDFS_PORT.to_string(), "9000".to_string()), + ]), + |warning| warnings.push(warning), + ) + .unwrap(); + + assert_eq!(cfg.options, None, "ignored keys must not be forwarded"); + assert_eq!(warnings, [ + "`hdfs.user` is not supported by the hdfs-native backend and is ignored", + "`hdfs.kerberos_ticket` is not supported by the hdfs-native backend and is ignored", + "`hdfs.port` has no effect without `hdfs.host` and is ignored", + ]); + } + + #[test] + fn test_hdfs_native_config_parse_rejects_invalid_name_node_entries() { + let parse = |value: &str| { + hdfs_native_config_parse(HashMap::from([( + HDFS_NAME_NODE.to_string(), + value.to_string(), + )])) + }; + + // Logical names, other schemes, userinfo, paths, port 0, unbracketed + // IPv6 and malformed ports: every entry must be a dialable host:port. + for value in [ + "hdfs://ns1", + "nn", + "nn:x", + "nn:0", + "nn:+80", + "::1", + "hdfs://[::1]", + "viewfs://nn:8020", + "foo://nn:8020", + "user@nn:8020", + "nn:8020/path", + "hdfs://nn1:8020,nn2", + ] { + let err = parse(value).unwrap_err().to_string(); + assert!(err.contains(HDFS_NAME_NODE), "{value}: {err}"); + } + } + + #[test] + fn test_hdfs_native_default_fs_must_be_hdfs() { + let parse = |props: &[(&str, &str)]| { + hdfs_native_config_parse( + props + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + ) + }; + let default_fs = |value| parse(&[("hadoop.fs.defaultFS", value)]); + + let (nn, _) = + hdfs_native_effective_name_node(&default_fs("hdfs://nn:8020/").unwrap(), "hdfs:///a") + .unwrap(); + assert_eq!(nn, "hdfs://nn:8020"); + // Not an HDFS `host:port`, including the portless logical-name form, + // which only a declaration resolves: rejected when the config is + // parsed, since it serves every authority-less path. + for value in ["viewfs://cluster/", "hdfs://", "hdfs://ns1"] { + let err = default_fs(value).unwrap_err(); + assert!( + err.to_string().contains(FS_DEFAULT_FS) && err.to_string().contains(HDFS_NAME_NODE), + "{value}: {err}" + ); + } + // Not checked when `hdfs.name-node` serves those paths instead. + parse(&[ + ("hadoop.fs.defaultFS", "viewfs://cluster/"), + (HDFS_NAME_NODE, "nn:8020"), + ]) + .unwrap(); + // Empty is unset, so the authority-less path has nothing to resolve to. + let err = + hdfs_native_effective_name_node(&default_fs("").unwrap(), "hdfs:///a").unwrap_err(); + assert!(err.to_string().contains(HDFS_HOST), "{err}"); + // Normalized like every other source. + let (nn, _) = + hdfs_native_effective_name_node(&default_fs("nn:9000").unwrap(), "hdfs:///a").unwrap(); + assert_eq!(nn, "hdfs://nn:9000"); + } + + /// A bare configured NameNode and the same NameNode as a path authority + /// share one cache entry. + #[test] + fn test_hdfs_native_create_operator_dedupes_spellings() { + let config = Arc::new( + hdfs_native_config_parse(HashMap::from([( + HDFS_NAME_NODE.to_string(), + "nn:8020".to_string(), + )])) + .unwrap(), + ); + let operators = HdfsNativeOperatorCache::default(); + + for path in ["hdfs:///a", "hdfs://nn:8020/b", "hdfs://NN:8020/c"] { + hdfs_native_create_operator(path, &config, &operators).unwrap(); + } + + assert_eq!(operators.len(), 1); + assert!(operators.get("hdfs://nn:8020").is_some()); + } + + #[test] + fn test_hdfs_native_config_parse_declares_nameservices() { + let cfg = hdfs_native_config_parse(HashMap::from([ + ( + format!("{HDFS_NAME_NODE}.ns-a"), + "nn1:8020, hdfs://nn2:8020/".to_string(), + ), + ("hadoop.dfs.ha.namenodes.ns-b".to_string(), "x".to_string()), + ( + "hadoop.dfs.namenode.rpc-address.ns-b.x".to_string(), + "nn3:8020".to_string(), + ), + ])) + .unwrap(); + let options = cfg.options.clone().unwrap(); + + // The sugar expands to Hadoop's keys... + assert_eq!( + options.get("dfs.ha.namenodes.ns-a").map(String::as_str), + Some("nn0,nn1") + ); + assert_eq!( + options + .get("dfs.namenode.rpc-address.ns-a.nn0") + .map(String::as_str), + Some("nn1:8020") + ); + assert_eq!( + options + .get("dfs.namenode.rpc-address.ns-a.nn1") + .map(String::as_str), + Some("nn2:8020") + ); + // ...and the resolver reads both forms back. + assert_eq!( + hdfs_native_nameservice(&cfg, "ns-a").unwrap().as_deref(), + Some("hdfs://nn1:8020,hdfs://nn2:8020") + ); + assert_eq!( + hdfs_native_nameservice(&cfg, "ns-b").unwrap().as_deref(), + Some("hdfs://nn3:8020") + ); + assert_eq!(hdfs_native_nameservice(&cfg, "ns-c").unwrap(), None); + + for (key, value) in [ + (format!("{HDFS_NAME_NODE}."), "nn:8020"), + (format!("{HDFS_NAME_NODE}.ns"), "nn"), + (format!("{HDFS_NAME_NODE}.ns"), " "), + ] { + let err = hdfs_native_config_parse(HashMap::from([(key.clone(), value.to_string())])) + .unwrap_err(); + assert!(err.to_string().contains(&key), "{key}={value}: {err}"); + } + // A passed-through declaration with a missing address fails when the + // config is parsed, not at the first path that names it. + let err = hdfs_native_config_parse(HashMap::from([( + "hadoop.dfs.ha.namenodes.ns-d".to_string(), + "a".to_string(), + )])) + .unwrap_err(); + assert!( + err.to_string().contains("dfs.namenode.rpc-address.ns-d.a"), + "{err}" + ); + } + + #[test] + fn test_hdfs_native_effective_name_node_declared_nameservices() { + let parse = |props: &[(String, &str)]| { + hdfs_native_config_parse( + props + .iter() + .map(|(k, v)| (k.clone(), v.to_string())) + .collect(), + ) + .unwrap() + }; + let ns_a = ( + format!("{HDFS_NAME_NODE}.ns-a"), + "hdfs://a1:8020,hdfs://a2:8020", + ); + let declared_only = parse(std::slice::from_ref(&ns_a)); + let with_default = parse(&[ns_a.clone(), (HDFS_NAME_NODE.to_string(), "hdfs://d:8020")]); + + // A declared nameservice resolves to its own list, ahead of the plain key. + for config in [&declared_only, &with_default] { + let (nn, rel) = hdfs_native_effective_name_node(config, "hdfs://ns-a/x").unwrap(); + assert_eq!((nn.as_str(), rel), ("hdfs://a1:8020,hdfs://a2:8020", "x")); + } + // An undeclared one is an error, with or without the plain key. + for config in [&declared_only, &with_default] { + let err = hdfs_native_effective_name_node(config, "hdfs://ns-b/x").unwrap_err(); + assert!(err.to_string().contains("`hdfs.name-node.ns-b`"), "{err}"); + } + // The plain key serves authority-less paths only. + let (nn, _) = hdfs_native_effective_name_node(&with_default, "hdfs:///x").unwrap(); + assert_eq!(nn, "hdfs://d:8020"); + // A concrete authority is never redirected. + let (nn, _) = + hdfs_native_effective_name_node(&with_default, "hdfs://other:9000/x").unwrap(); + assert_eq!(nn, "hdfs://other:9000"); + } + + #[test] + fn test_hdfs_native_create_operator_per_nameservice() { + let config = Arc::new( + hdfs_native_config_parse(HashMap::from([ + ( + format!("{HDFS_NAME_NODE}.ns-a"), + "hdfs://a:8020".to_string(), + ), + ( + format!("{HDFS_NAME_NODE}.ns-b"), + "hdfs://b:8020".to_string(), + ), + (HDFS_NAME_NODE.to_string(), "hdfs://d:8020".to_string()), + ])) + .unwrap(), + ); + let operators = HdfsNativeOperatorCache::default(); + + for path in ["hdfs://ns-a/x", "hdfs://ns-b/y", "hdfs:///w"] { + hdfs_native_create_operator(path, &config, &operators).unwrap(); + } + // ns-a, ns-b, and the plain key for the authority-less path; an + // undeclared nameservice never reaches an operator. + assert_eq!(operators.len(), 3); + assert_eq!( + hdfs_native_batch_key(&config, "hdfs://ns-b/y"), + "hdfs://b:8020" + ); + let err = hdfs_native_create_operator("hdfs://ns-c/z", &config, &operators).unwrap_err(); + assert!(err.to_string().contains("`hdfs.name-node.ns-c`"), "{err}"); + assert_eq!(operators.len(), 3); + } + + #[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(), + ), + (HDFS_HOST.to_string(), "ignored".to_string()), + ])) + .unwrap(); + let default_fs = + hdfs_native_config_parse(HashMap::from([(HDFS_HOST.to_string(), "nn".to_string())])) + .unwrap(); + let unconfigured = HdfsNativeConfig::default(); + + // An authority with a port is used as is, whatever is configured. + for config in [&configured, &default_fs, &unconfigured] { + let (nn, rel) = + hdfs_native_effective_name_node(config, "hdfs://other:9000/a/b").unwrap(); + assert_eq!((nn.as_str(), rel), ("hdfs://other:9000", "a/b")); + } + // An authority-less path uses the configured list, ahead of `hdfs.host`. + let (nn, _) = hdfs_native_effective_name_node(&configured, "hdfs:///y").unwrap(); + assert_eq!(nn, "hdfs://nn1:8020,hdfs://nn2:8020"); + // A logical nameservice never falls back to it: it must be declared. + let err = hdfs_native_effective_name_node(&configured, "hdfs://ns-a/x").unwrap_err(); + assert!(err.to_string().contains("`hdfs.name-node.ns-a`"), "{err}"); + // Only an authority-less path falls back to `fs.defaultFS`. + let (nn, rel) = hdfs_native_effective_name_node(&default_fs, "hdfs:///y").unwrap(); + assert_eq!((nn.as_str(), rel), ("hdfs://nn:8020", "y")); + let err = hdfs_native_effective_name_node(&default_fs, "hdfs://ns-a/x").unwrap_err(); + assert!(err.to_string().contains("`hdfs.name-node.ns-a`"), "{err}"); + // Nothing applicable: pointed errors. + for path in ["hdfs:///a", "hdfs://ns-a/x"] { + let err = hdfs_native_effective_name_node(&unconfigured, path).unwrap_err(); + assert!(err.to_string().contains(HDFS_NAME_NODE), "{err}"); + } + } + + #[test] + 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, + HdfsNativeAuthority::Nameservice("nameservice1".to_string()) + ); + assert_eq!(rel, "a/b"); + } + + #[test] + fn test_hdfs_native_parse_path_with_authority_and_port() { + let (nn, rel) = hdfs_native_parse_path("hdfs://nn:8020/foo").unwrap(); + // Extra leading slashes go too: opendal rejects paths starting with `/`. + assert_eq!( + hdfs_native_parse_path("hdfs://nn:8020//foo").unwrap().1, + "foo" + ); + + assert_eq!( + nn, + HdfsNativeAuthority::NameNode("hdfs://nn:8020".to_string()) + ); + assert_eq!(rel, "foo"); + } + + #[test] + fn test_hdfs_native_parse_path_ipv6_authority_keeps_brackets() { + let (nn, rel) = hdfs_native_parse_path("hdfs://[::1]:8020/a/b").unwrap(); + assert_eq!( + nn, + HdfsNativeAuthority::NameNode("hdfs://[::1]:8020".to_string()) + ); + assert_eq!(rel, "a/b"); + } + + #[test] + fn test_hdfs_native_parse_path_rejects_userinfo_and_port_zero() { + for (path, reason) in [ + ("hdfs://alice@nn:8020/a", "userinfo"), + ("hdfs://alice:secret@nn:8020/a", "userinfo"), + ("hdfs://nn:0/a", "port 0"), + ] { + let err = hdfs_native_parse_path(path).unwrap_err().to_string(); + assert!(err.contains(reason), "{path}: {err}"); + // A password must never reach a log. + assert!(!err.contains("alice") && !err.contains("secret"), "{err}"); + } + } + + #[test] + fn test_hdfs_native_errors_mask_userinfo() { + let path_err = |path: &str| { + hdfs_native_effective_name_node(&HdfsNativeConfig::default(), path) + .unwrap_err() + .to_string() + }; + let config_err = |key: &str, value: &str| { + let props = HashMap::from([(key.to_string(), value.to_string())]); + hdfs_native_config_parse(props) + .and_then(|config| hdfs_native_effective_name_node(&config, "hdfs:///x")) + .unwrap_err() + .to_string() + }; + // Every error that echoes a path or a NameNode value, including the + // ones that fail before the userinfo check can run. + for (err, masked) in [ + ( + path_err("hdfs://alice:s3cret@nn:bad/x"), + "hdfs://***@nn:bad/x", + ), + (path_err("s3://alice:s3cret@b/x"), "s3://***@b/x"), + ( + path_err("hdfs://alice:s3cret@nn:8020/x"), + "hdfs://***@nn:8020/x", + ), + ( + config_err(HDFS_NAME_NODE, "hdfs://alice:s3cret@nn:8020"), + "hdfs://***@nn:8020", + ), + ( + config_err("hdfs.name-node.ns", "hdfs://alice:s3cret@nn:8020"), + "hdfs://***@nn:8020", + ), + (config_err(HDFS_HOST, "alice:s3cret@nn"), "***@nn:8020"), + ( + config_err("hadoop.fs.defaultFS", "hdfs://alice:s3cret@nn:8020"), + "hdfs://***@nn:8020", + ), + // A raw `/`, `#` or `://` in the password, or a second `@`, must + // not move where the masking stops. + ( + path_err("hdfs://alice:s3/cret@nn:8020/x"), + "hdfs://***@nn:8020/x", + ), + (path_err("s3://alice:12#cret@b/x"), "s3://***@b/x"), + ( + path_err("hdfs://alice@corp:s3cret@nn:bad/x"), + "hdfs://***@nn:bad/x", + ), + ( + config_err(HDFS_NAME_NODE, "hdfs://alice:s3/cret@nn:8020"), + "hdfs://***@nn:8020", + ), + (config_err(HDFS_HOST, "alice:s3://cret@nn"), "***@nn:8020"), + ] { + assert!(err.contains(masked), "{err}"); + assert!(!err.contains("alice") && !err.contains("cret"), "{err}"); + } + } + + #[test] + fn test_hdfs_native_parse_path_with_authority_no_path() { + let (nn, rel) = hdfs_native_parse_path("hdfs://nameservice1").unwrap(); + + assert_eq!( + nn, + HdfsNativeAuthority::Nameservice("nameservice1".to_string()) + ); + assert_eq!(rel, ""); + } + + #[test] + fn test_hdfs_native_parse_path_with_authority_trailing_slash() { + let (nn, rel) = hdfs_native_parse_path("hdfs://nameservice1/").unwrap(); + + assert_eq!( + nn, + HdfsNativeAuthority::Nameservice("nameservice1".to_string()) + ); + assert_eq!(rel, ""); + } + + #[test] + fn test_hdfs_native_parse_path_authority_less_returns_none() { + let (nn, rel) = hdfs_native_parse_path("hdfs:///a/b").unwrap(); + + assert_eq!(nn, HdfsNativeAuthority::Default); + assert_eq!(rel, "a/b"); + } + + #[test] + 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_hdfs_native_parse_path_invalid_url_errors() { + let err = hdfs_native_parse_path("not-a-url").unwrap_err().to_string(); + + // The URL parser's reason is forwarded; this is not the scheme check. + assert!( + err.contains("Invalid hdfs path") && !err.contains("expected scheme"), + "{err}" + ); + } + + #[test] + 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 = hdfs_native_parse_path(path).unwrap_err(); + assert!(err.to_string().contains("expected scheme `hdfs://`")); + } + } + + #[test] + fn test_hdfs_native_batch_key_distinguishes_ports() { + let config = HdfsNativeConfig::default(); + + assert_eq!( + hdfs_native_batch_key(&config, "hdfs://namenode:8020/a"), + "hdfs://namenode:8020" + ); + assert_eq!( + hdfs_native_batch_key(&config, "hdfs://namenode:9000/b"), + "hdfs://namenode:9000" + ); + } + + #[test] + fn test_hdfs_native_batch_key_invalid_path_keys_on_itself() { + let config = HdfsNativeConfig::default(); + + // Unresolvable paths must not collapse onto a shared "" key. + for path in ["not-a-url", "hdfs:///a", "hdfs://nn:0/a"] { + assert_eq!(hdfs_native_batch_key(&config, path), path); + } + } + + // Creating an operator offloads the build to a blocking thread, so these + // run under tokio. + /// `relativize_path` follows the same resolution rule as `create_operator`. + #[test] + fn test_hdfs_native_relativize_path() { + let build = |key: String| { + let props = HashMap::from([(key, "hdfs://nn:8020".to_string())]); + crate::OpenDalStorage::HdfsNative(HdfsNativeStorage::new( + hdfs_native_config_parse(props).unwrap(), + OpenDalClientConfig::default(), + )) + }; + let unconfigured = crate::OpenDalStorage::HdfsNative(HdfsNativeStorage::new( + HdfsNativeConfig::default(), + OpenDalClientConfig::default(), + )); + let declared = build(format!("{HDFS_NAME_NODE}.nameservice1")); + let plain = build(HDFS_NAME_NODE.to_string()); + + // A concrete authority needs nothing configured. + assert_eq!( + unconfigured + .relativize_path("hdfs://nn:8020/warehouse/db/t") + .unwrap(), + "warehouse/db/t" + ); + // A logical nameservice needs its declaration, an authority-less path + // the plain key. + for (storage, path, rel) in [ + (&declared, "hdfs://nameservice1/a/b.parquet", "a/b.parquet"), + (&plain, "hdfs:///a/b", "a/b"), + ] { + let err = unconfigured.relativize_path(path).unwrap_err(); + assert!(err.to_string().contains(HDFS_NAME_NODE), "{err}"); + assert_eq!(storage.relativize_path(path).unwrap(), rel); + } + // Another scheme is rejected before any resolution. + let err = unconfigured.relativize_path("s3://bucket/x").unwrap_err(); + assert!( + err.to_string().contains("expected scheme `hdfs://`"), + "{err}" + ); + } + + #[test] + fn test_hdfs_native_create_operator_plain_key_serves_authority_less_paths() { + let config = Arc::new( + hdfs_native_config_parse(HashMap::from([( + HDFS_NAME_NODE.to_string(), + "hdfs://nn1:8020/,hdfs://nn2:8020/".to_string(), + )])) + .unwrap(), + ); + let operators = HdfsNativeOperatorCache::default(); + + // Authority-less paths share the configured list's operator. + let (_, rel) = hdfs_native_create_operator("hdfs:///a/b", &config, &operators).unwrap(); + assert_eq!(rel, "a/b"); + assert_eq!(operators.len(), 1); + assert!(operators.get("hdfs://nn1:8020,hdfs://nn2:8020").is_some()); + + // A logical nameservice must be declared; a concrete authority is + // another cluster, never the configured one. + let err = hdfs_native_create_operator("hdfs://ns-a/c", &config, &operators).unwrap_err(); + assert!(err.to_string().contains("`hdfs.name-node.ns-a`"), "{err}"); + hdfs_native_create_operator("hdfs://other:9000/d", &config, &operators).unwrap(); + assert_eq!(operators.len(), 2); + assert!(operators.get("hdfs://other:9000").is_some()); + } + + #[test] + fn test_hdfs_native_create_operator_caches_per_name_node() { + let config = Arc::new(HdfsNativeConfig::default()); + let operators = HdfsNativeOperatorCache::default(); + + for path in [ + "hdfs://nn1:8020/a", + "hdfs://nn1:8020/b", + "hdfs://nn2:8020/c", + ] { + hdfs_native_create_operator(path, &config, &operators).unwrap(); + } + + assert_eq!(operators.len(), 2); + } + + #[tokio::test] + async fn test_hdfs_native_operator_cache_keeps_first_insert() { + let operators = HdfsNativeOperatorCache::default(); + let build = |root: &str| { + let mut config = hdfs_native_config_parse(HashMap::new()).unwrap(); + config.root = Some(root.to_string()); + hdfs_native_operator_build(&config, "hdfs://nn:8020").unwrap() + }; + + // Two callers racing for one NameNode: the first insert wins and both + // get the cached operator. + let first = operators.insert("hdfs://nn:8020".to_string(), build("/first")); + let second = operators.insert("hdfs://nn:8020".to_string(), build("/second")); + assert_eq!(first.info().root(), "/first/"); + assert_eq!(second.info().root(), "/first/"); + assert_eq!(operators.len(), 1); + } + + #[test] + fn test_hdfs_native_operator_cache_survives_a_poisoned_lock() { + let operators = HdfsNativeOperatorCache::default(); + // A panic elsewhere while the lock is held must not break HDFS I/O + // for good: every write inserts a whole entry, so the map is intact. + let _ = std::panic::catch_unwind(|| { + let _guard = operators.0.write().unwrap(); + panic!("poisoning the operator cache lock"); + }); + assert!(operators.0.is_poisoned()); + + let config = Arc::new(HdfsNativeConfig::default()); + hdfs_native_create_operator("hdfs://nn:8020/a", &config, &operators).unwrap(); + assert!(operators.get("hdfs://nn:8020").is_some()); + } + + /// The configuration round-trips through serde; the operator cache does + /// not and starts empty. + #[test] + fn test_hdfs_native_storage_serde_round_trip() { + let props = HashMap::from([ + (HDFS_NAME_NODE.to_string(), "hdfs://nn:8020".to_string()), + ( + "hadoop.dfs.client.use.datanode.hostname".to_string(), + "true".to_string(), + ), + ]); + let storage = crate::OpenDalStorage::HdfsNative(HdfsNativeStorage::new( + hdfs_native_config_parse(props).unwrap(), + OpenDalClientConfig::from_properties(&HashMap::from([( + crate::OPENDAL_IO_TIMEOUT_MS.to_string(), + "45000".to_string(), + )])) + .unwrap(), + )); + // Populate the cache so that its absence after the round trip means + // something. + let crate::OpenDalStorage::HdfsNative(hdfs) = &storage else { + panic!("expected the HdfsNative variant"); + }; + hdfs_native_create_operator("hdfs:///x", &hdfs.config, &hdfs.operators).unwrap(); + assert!(hdfs.operators.get("hdfs://nn:8020").is_some()); + + let value = serde_json::to_value(&storage).unwrap(); + assert!(value["HdfsNative"].get("operators").is_none()); + let crate::OpenDalStorage::HdfsNative(restored) = serde_json::from_value(value).unwrap() + else { + panic!("expected the HdfsNative variant"); + }; + assert_eq!(restored.config.name_node.as_deref(), Some("hdfs://nn:8020")); + assert_eq!( + restored + .config + .options + .as_ref() + .and_then(|o| o.get("dfs.client.use.datanode.hostname")), + Some(&"true".to_string()) + ); + assert_eq!( + restored.client_config.io_timeout(), + std::time::Duration::from_secs(45) + ); + assert!(restored.operators.get("hdfs://nn:8020").is_none()); + } +} diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 9e01552016..693f4c18b3 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -77,6 +77,14 @@ cfg_if! { } } +cfg_if! { + if #[cfg(feature = "opendal-hdfs-native")] { + mod hdfs_native; + use hdfs_native::*; + pub use hdfs_native::HdfsNativeStorage; + } +} + cfg_if! { if #[cfg(feature = "opendal-memory")] { mod memory; @@ -186,6 +194,9 @@ pub enum OpenDalStorageFactory { /// GCS storage factory. #[cfg(feature = "opendal-gcs")] Gcs, + /// HDFS storage factory. + #[cfg(feature = "opendal-hdfs-native")] + HdfsNative, /// OSS storage factory. #[cfg(feature = "opendal-oss")] Oss, @@ -236,6 +247,13 @@ impl StorageFactory for OpenDalStorageFactory { config: gcs_config_parse(config.props().clone())?.into(), client_config, })), + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorageFactory::HdfsNative => Ok(Arc::new(OpenDalStorage::HdfsNative( + HdfsNativeStorage::new( + hdfs_native_config_parse(config.props().clone())?, + client_config, + ), + ))), #[cfg(feature = "opendal-oss")] OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::Oss { config: oss_config_parse(config.props().clone())?.into(), @@ -259,6 +277,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, @@ -320,6 +339,9 @@ pub enum OpenDalStorage { #[serde(default)] client_config: OpenDalClientConfig, }, + /// HDFS storage variant; see [`HdfsNativeStorage`]. + #[cfg(feature = "opendal-hdfs-native")] + HdfsNative(HdfsNativeStorage), /// OSS storage variant. #[cfg(feature = "opendal-oss")] Oss { @@ -435,6 +457,10 @@ impl OpenDalStorage { )); } } + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorage::HdfsNative(storage) => { + hdfs_native_create_operator(path, &storage.config, &storage.operators)? + } #[cfg(feature = "opendal-oss")] OpenDalStorage::Oss { config, .. } => { let op = oss_config_build(config, path)?; @@ -459,6 +485,7 @@ impl OpenDalStorage { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs-native"), ))] _ => { return Err(Error::new( @@ -500,6 +527,8 @@ impl OpenDalStorage { OpenDalStorage::Azdls { client_config, .. } => client_config, #[cfg(feature = "opendal-hf")] OpenDalStorage::Hf { client_config, .. } => client_config, + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorage::HdfsNative(storage) => &storage.client_config, #[cfg(all( not(feature = "opendal-memory"), not(feature = "opendal-s3"), @@ -508,6 +537,7 @@ impl OpenDalStorage { not(feature = "opendal-oss"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), + not(feature = "opendal-hdfs-native"), ))] _ => unreachable!(), } @@ -521,6 +551,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::HdfsNative(storage) => hdfs_native_batch_key(&storage.config, path), _ => url::Url::parse(path) .ok() .and_then(|u| u.host_str().map(|s| s.to_string())) @@ -580,6 +614,11 @@ impl OpenDalStorage { )) } } + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorage::HdfsNative(storage) => { + let (_, relative_path) = hdfs_native_effective_name_node(&storage.config, path)?; + Ok(relative_path) + } #[cfg(feature = "opendal-oss")] OpenDalStorage::Oss { .. } => { let url = url::Url::parse(path)?; @@ -619,6 +658,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, @@ -929,6 +969,8 @@ mod tests { OpenDalStorageFactory::Azdls, #[cfg(feature = "opendal-hf")] OpenDalStorageFactory::Hf, + #[cfg(feature = "opendal-hdfs-native")] + OpenDalStorageFactory::HdfsNative, ]; for factory in factories { // `build` returns `dyn Storage`, so read the config back from its serialized form. diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 51e76da472..d7a977e7ac 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}"), @@ -136,6 +138,14 @@ fn build_storage_for_scheme( client_config, }) } + #[cfg(feature = "opendal-hdfs-native")] + "hdfs" => { + let config = crate::hdfs_native::hdfs_native_config_parse(props.clone())?; + Ok(OpenDalStorage::HdfsNative(crate::HdfsNativeStorage::new( + config, + client_config, + ))) + } unsupported => Err(Error::new( ErrorKind::FeatureUnsupported, format!("Unsupported storage scheme: {unsupported}"), @@ -438,6 +448,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_native_returns_hdfs_variant() { + let storage = empty_resolving_storage(); + + let resolved = storage.resolve("hdfs://nameservice1/a/b").unwrap(); + + assert!( + matches!(&*resolved, OpenDalStorage::HdfsNative(_)), + "expected HdfsNative variant, got {resolved:?}" + ); + } + + #[cfg(feature = "opendal-hdfs-native")] + #[test] + fn test_resolve_hdfs_native_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::HdfsNative 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..ef9da20a13 --- /dev/null +++ b/crates/storage/opendal/tests/file_io_hdfs_test.rs @@ -0,0 +1,400 @@ +// 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 fixture in `dev/docker-compose.yaml`, which +//! `make docker-up` starts with the other fixtures. + +#[cfg(feature = "opendal-hdfs-native")] +mod tests { + use std::net::ToSocketAddrs; + use std::sync::Arc; + + use bytes::Bytes; + use futures::StreamExt; + use iceberg::io::{ + FileIO, FileIOBuilder, HDFS_HOST, HDFS_NAME_NODE, HDFS_PORT, StorageFactory, + }; + use iceberg_storage_opendal::{OpenDalResolvingStorageFactory, OpenDalStorageFactory}; + use iceberg_test_utils::{get_hdfs_endpoint, normalize_test_name_with_parts, set_up}; + + /// The fixture's DataNode advertises `127.0.0.1`; without this the client + /// dials its container IP, which the host cannot reach on macOS or Windows. + fn file_io_builder(factory: Arc) -> FileIOBuilder { + set_up(); + FileIOBuilder::new(factory).with_prop("hadoop.dfs.client.use.datanode.hostname", "true") + } + + fn get_file_io() -> FileIO { + file_io_builder(Arc::new(OpenDalStorageFactory::HdfsNative)).build() + } + + fn test_path(suffix: &str) -> String { + format!( + "{}/{}", + get_hdfs_endpoint(), + normalize_test_name_with_parts!(suffix) + ) + } + + /// The endpoint with its host replaced by that host's IPv4 address: a + /// second spelling of the same NameNode, so a second batch key. `None` + /// when the endpoint already uses an IP literal. + fn alternate_endpoint(endpoint: &str) -> Option { + let url = url::Url::parse(endpoint).ok()?; + let host = url.host_str()?; + if host.parse::().is_ok() { + return None; + } + let addr = (host, url.port()?) + .to_socket_addrs() + .ok()? + .find(std::net::SocketAddr::is_ipv4)?; + Some(format!("hdfs://{addr}")) + } + + #[tokio::test] + async fn test_file_io_hdfs_exists() { + let file_io = get_file_io(); + let dir = test_path("test_file_io_hdfs_exists"); + let present = format!("{dir}/present"); + let _ = file_io.delete(&present).await; + file_io + .new_output(&present) + .unwrap() + .write(Bytes::from_static(b"x")) + .await + .unwrap(); + + assert!(file_io.exists(&present).await.unwrap()); + assert!(!file_io.exists(&format!("{dir}/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, and + /// `hdfs.name-node.` declares its comma-separated endpoints. + /// The first one refuses connections, so every call must fail over to + /// the fixture through opendal's synthetic nameservice. + #[tokio::test] + async fn test_file_io_hdfs_declared_nameservice() { + let endpoint = get_hdfs_endpoint(); + // A local port nothing listens on once the listener is dropped. + let dead = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap(); + let file_io = file_io_builder(Arc::new(OpenDalStorageFactory::HdfsNative)) + .with_prop( + format!("{HDFS_NAME_NODE}.logical-nameservice"), + format!("hdfs://{dead},{endpoint}"), + ) + .build(); + + // The path authority is a logical name resolved by its declaration. + let path = format!( + "hdfs://logical-nameservice/{}", + normalize_test_name_with_parts!("test_file_io_hdfs_declared_nameservice") + ); + 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") + ); + } + + /// Authority-less paths resolve through plain `hdfs.name-node`, or through + /// PyIceberg's `hdfs.host`/`hdfs.port`; both must reach the same cluster. + #[tokio::test] + async fn test_file_io_hdfs_authority_less_paths() { + let endpoint = get_hdfs_endpoint(); + let url = url::Url::parse(&endpoint).unwrap(); + let by_name_node = file_io_builder(Arc::new(OpenDalStorageFactory::HdfsNative)) + .with_prop(HDFS_NAME_NODE, &endpoint) + .build(); + let by_host_port = file_io_builder(Arc::new(OpenDalStorageFactory::HdfsNative)) + .with_prop(HDFS_HOST, url.host_str().unwrap()) + .with_prop(HDFS_PORT, url.port().unwrap().to_string()) + .build(); + + let path = format!( + "hdfs:///{}", + normalize_test_name_with_parts!("test_file_io_hdfs_authority_less_paths") + ); + let _ = by_name_node.delete(&path).await; + by_name_node + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"authority-less")) + .await + .unwrap(); + + assert_eq!( + by_host_port.new_input(&path).unwrap().read().await.unwrap(), + Bytes::from_static(b"authority-less") + ); + by_host_port.delete(&path).await.unwrap(); + assert!(!by_name_node.exists(&path).await.unwrap()); + } + + #[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()); + } + } + + /// Paths are batched per effective NameNode; two spellings of the fixture's + /// NameNode drive two operators and two deleters through one stream. The + /// fixture is one cluster, so the keying itself is pinned by the unit + /// tests; this proves the multi-operator path completes and cleans up. + #[tokio::test] + async fn test_file_io_hdfs_delete_stream_two_name_nodes() { + let endpoint = get_hdfs_endpoint(); + let Some(alternate) = alternate_endpoint(&endpoint) else { + eprintln!("Skipping HDFS test: {endpoint} has no second spelling"); + return; + }; + let file_io = get_file_io(); + let dir = test_path("test_file_io_hdfs_delete_stream_two_name_nodes"); + let alt_dir = dir.replacen(&endpoint, &alternate, 1); + assert_ne!(dir, alt_dir); + + let paths = vec![format!("{dir}/a"), format!("{alt_dir}/b")]; + for path in &paths { + let _ = file_io.delete(path).await; + file_io + .new_output(path) + .unwrap() + .write("delete-me".into()) + .await + .unwrap(); + } + // Both spellings reach the same cluster. + assert!(file_io.exists(&format!("{alt_dir}/a")).await.unwrap()); + assert!(file_io.exists(&format!("{dir}/b")).await.unwrap()); + + let stream = futures::stream::iter(paths.clone()).boxed(); + file_io.delete_stream(stream).await.unwrap(); + + // Everything is gone, through whichever deleter. + for path in [ + format!("{dir}/a"), + format!("{dir}/b"), + format!("{alt_dir}/a"), + format!("{alt_dir}/b"), + ] { + assert!(!file_io.exists(&path).await.unwrap(), "{path}"); + } + } + + #[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() { + let file_io = file_io_builder(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/storage/opendal/tests/hdfs_native_runtime_test.rs b/crates/storage/opendal/tests/hdfs_native_runtime_test.rs new file mode 100644 index 0000000000..da22260d72 --- /dev/null +++ b/crates/storage/opendal/tests/hdfs_native_runtime_test.rs @@ -0,0 +1,91 @@ +// 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. + +//! A cached HDFS operator must keep working after the runtime that first +//! used it is dropped. Needs no HDFS: the NameNode is a local listener that +//! only counts dials. + +#[cfg(feature = "opendal-hdfs-native")] +mod tests { + use std::io::Read; + use std::net::{TcpListener, TcpStream}; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + use iceberg::io::{FileIO, FileIOBuilder, HDFS_NAME_NODE}; + use iceberg_storage_opendal::OpenDalStorageFactory; + + /// Accepts and immediately closes connections, counting them. A failed + /// accept (a dial reset while queued) is skipped so the listener lives on. + fn fake_name_node() -> (u16, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let dials = Arc::new(AtomicUsize::new(0)); + let counter = dials.clone(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(_stream) = stream else { continue }; + counter.fetch_add(1, Ordering::SeqCst); + } + }); + (port, dials) + } + + fn stat(runtime: &tokio::runtime::Runtime, file_io: &FileIO) { + runtime.block_on(async { + let input = file_io.new_input("hdfs:///f").unwrap(); + // The fake NameNode never answers, so this fails; only the dial matters. + let _ = tokio::time::timeout(Duration::from_secs(10), input.metadata()).await; + }); + } + + #[test] + fn test_hdfs_operator_outlives_the_runtime_that_first_used_it() { + let (port, dials) = fake_name_node(); + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::HdfsNative)) + .with_prop(HDFS_NAME_NODE, format!("hdfs://127.0.0.1:{port}")) + .with_prop("hadoop.dfs.client.failover.max.attempts", "1") + .build(); + + let first = tokio::runtime::Runtime::new().unwrap(); + stat(&first, &file_io); + assert!( + dials.load(Ordering::SeqCst) >= 1, + "the NameNode was never dialed" + ); + drop(first); + + // The listener accepts in order, so once it has closed this marker + // connection, every dial from the first runtime has been counted. + let mut marker = TcpStream::connect(("127.0.0.1", port)).unwrap(); + marker + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + assert_eq!(marker.read(&mut [0]).unwrap(), 0); + let after_first = dials.load(Ordering::SeqCst); + + // Reusing the FileIO from another runtime used to panic inside + // hdfs-native, whose client was bound to the dropped runtime. + let second = tokio::runtime::Runtime::new().unwrap(); + stat(&second, &file_io); + assert!( + dials.load(Ordering::SeqCst) > after_first, + "no dial from the second runtime" + ); + } +} diff --git a/crates/test_utils/src/lib.rs b/crates/test_utils/src/lib.rs index bf2dad0ed6..176767813e 100644 --- a/crates/test_utils/src/lib.rs +++ b/crates/test_utils/src/lib.rs @@ -47,6 +47,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_OBJECT_STORE_PORT: u16 = 9000; @@ -54,6 +55,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_PORT: u16 = 8020; /// Returns the S3-compatible object store endpoint. /// Checks ICEBERG_TEST_OBJECT_STORE_ENDPOINT env var, otherwise returns localhost default. @@ -89,6 +91,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_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 cb89e3042d..3bf9d730b5 100644 --- a/dev/docker-compose.yaml +++ b/dev/docker-compose.yaml @@ -137,6 +137,51 @@ services: timeout: 5s retries: 5 + # ============================================================================= + # HDFS - single-node NameNode + DataNode for HDFS tests + # ============================================================================= + # The DataNode advertises 127.0.0.1, so clients that set + # `dfs.client.use.datanode.hostname` reach it through its published port. + hdfs-namenode: + image: apache/hadoop:3.5.0 + networks: + iceberg_test: + command: ["hdfs", "namenode"] + environment: + ENSURE_NAMENODE_DIR: "/tmp/hadoop-root/dfs/name" + ports: + - "8020:8020" + 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 + networks: + iceberg_test: + command: ["hdfs", "datanode"] + depends_on: + hdfs-namenode: + condition: service_healthy + ports: + - "9866:9866" + 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..273676bba2 --- /dev/null +++ b/dev/hdfs/core-site.xml @@ -0,0 +1,27 @@ + + + + + fs.defaultFS + hdfs://hdfs-namenode: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..b9bf0c63c1 --- /dev/null +++ b/dev/hdfs/hdfs-site.xml @@ -0,0 +1,42 @@ + + + + + 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 + + + + dfs.datanode.hostname + 127.0.0.1 + +