comphead commented on code in PR #3111: URL: https://github.com/apache/iceberg-rust/pull/3111#discussion_r4018997879
########## crates/storage/opendal/src/hdfs_native.rs: ########## @@ -0,0 +1,350 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! HDFS storage backend via OpenDAL's `services-hdfs-native` (pure Rust, no JNI). + +use std::collections::HashMap; +use std::sync::RwLock; + +use iceberg::io::{HDFS_HADOOP_CONF_PREFIX, HDFS_NAME_NODE}; +use iceberg::{Error, ErrorKind, Result}; +use opendal::Operator; +use opendal::services::HdfsNativeConfig; +use url::Url; + +use crate::utils::from_opendal_error; + +/// Parse iceberg properties to [`HdfsNativeConfig`]. +pub(crate) fn hdfs_native_config_parse(mut m: HashMap<String, String>) -> Result<HdfsNativeConfig> { + let mut cfg = HdfsNativeConfig::default(); + + if let Some(name_node) = m.remove(HDFS_NAME_NODE) { + cfg.name_node = Some(name_node); + } + + let options: HashMap<String, String> = m + .into_iter() + .filter_map(|(key, value)| { + key.strip_prefix(HDFS_HADOOP_CONF_PREFIX) + .map(|stripped| (stripped.to_string(), value)) + }) + .collect(); + if !options.is_empty() { + cfg.options = Some(options); + } + + Ok(cfg) +} + +/// Parse an HDFS path into `Some("hdfs://<authority>")` (`None` when +/// authority-less) and the relative path (no leading `/`, opendal style). +pub(crate) fn hdfs_native_parse_path(path: &str) -> Result<(Option<String>, &str)> { + let url = Url::parse(path).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}: {e}"), + ) + })?; + // Non-special schemes parse even without `//` (e.g. `hdfs:x` is a valid + // non-hierarchical URL), so require the literal prefix before slicing. + let (Some(after_scheme), "hdfs") = (path.strip_prefix("hdfs://"), url.scheme()) else { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}, expected scheme `hdfs://`"), + )); + }; + + let name_node = url.host_str().filter(|h| !h.is_empty()).map(|host| { + url.port() + .map(|port| format!("hdfs://{host}:{port}")) + .unwrap_or_else(|| format!("hdfs://{host}")) + }); + + // `url.path()` borrows from `url` and can't be returned with the input's + // lifetime. Slice the path component out of the original input instead; + // it starts after the first `/` following the `hdfs://` prefix. Opendal + // paths must not start with `/` (`Deleter::delete` rejects them). + let rel = match after_scheme.find('/') { + Some(i) => after_scheme[i..].trim_start_matches('/'), + None => "", + }; + + Ok((name_node, rel)) +} + +/// Creates an operator for the path, cached per effective NameNode (the +/// configured `hdfs.name-node`, else the path authority) — each operator +/// holds an HDFS client with live RPC connections. +pub(crate) fn hdfs_native_create_operator<'a>( + path: &'a str, + config: &HdfsNativeConfig, + operators: &RwLock<HashMap<String, Operator>>, +) -> Result<(Operator, &'a str)> { + let (authority_name_node, relative_path) = hdfs_native_parse_path(path)?; + + let name_node = match config.name_node.clone().or(authority_name_node) { + Some(name_node) => name_node, + None => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid hdfs path: {path}, authority-less paths require the `{HDFS_NAME_NODE}` property" + ), + )); + } + }; + + // Fast path: check read lock first. + { + let cache = operators + .read() + .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; + if let Some(op) = cache.get(&name_node) { + return Ok((op.clone(), relative_path)); + } + } + + // Slow path: build and insert under write lock, re-checking for a + // concurrent insert. + let mut cache = operators + .write() + .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; + let op = match cache.get(&name_node) { + Some(op) => op.clone(), + None => { + let op = hdfs_native_operator_build(config, &name_node)?; Review Comment: **Blocker: blocking I/O runs while the cache write lock is held.** `hdfs_native_operator_build` -> `Operator::from_config` -> `HdfsNativeBuilder::build()` -> `hdfs_native::ClientBuilder::build()` -> `Configuration::new(..)`, which does synchronous `fs::read_to_string` on `core-site.xml`/`hdfs-site.xml` (hdfs-native 0.14.5, `src/common/config.rs:228`), followed by `NameServiceProxy::new`. All of that happens on a tokio worker thread, under `operators.write()`. Two effects: the worker thread is blocked on file I/O, and every concurrent caller is serialized behind this lock, including callers that would have hit the cache for a *different*, already-built NameNode. The double-checked-locking shape mirrors `OpenDalResolvingStorage::resolve`, but there the build is pure CPU. The cost profile is different here. Suggested: build outside the lock and insert with `entry(..).or_insert_with(..)`, accepting the occasional duplicate build, or move the build into `spawn_blocking`. ########## crates/storage/opendal/src/hdfs_native.rs: ########## @@ -0,0 +1,350 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! HDFS storage backend via OpenDAL's `services-hdfs-native` (pure Rust, no JNI). + +use std::collections::HashMap; +use std::sync::RwLock; + +use iceberg::io::{HDFS_HADOOP_CONF_PREFIX, HDFS_NAME_NODE}; +use iceberg::{Error, ErrorKind, Result}; +use opendal::Operator; +use opendal::services::HdfsNativeConfig; +use url::Url; + +use crate::utils::from_opendal_error; + +/// Parse iceberg properties to [`HdfsNativeConfig`]. +pub(crate) fn hdfs_native_config_parse(mut m: HashMap<String, String>) -> Result<HdfsNativeConfig> { + let mut cfg = HdfsNativeConfig::default(); + + if let Some(name_node) = m.remove(HDFS_NAME_NODE) { + cfg.name_node = Some(name_node); + } + + let options: HashMap<String, String> = m + .into_iter() + .filter_map(|(key, value)| { + key.strip_prefix(HDFS_HADOOP_CONF_PREFIX) + .map(|stripped| (stripped.to_string(), value)) + }) + .collect(); + if !options.is_empty() { + cfg.options = Some(options); + } + + Ok(cfg) +} + +/// Parse an HDFS path into `Some("hdfs://<authority>")` (`None` when +/// authority-less) and the relative path (no leading `/`, opendal style). +pub(crate) fn hdfs_native_parse_path(path: &str) -> Result<(Option<String>, &str)> { + let url = Url::parse(path).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}: {e}"), + ) + })?; + // Non-special schemes parse even without `//` (e.g. `hdfs:x` is a valid + // non-hierarchical URL), so require the literal prefix before slicing. + let (Some(after_scheme), "hdfs") = (path.strip_prefix("hdfs://"), url.scheme()) else { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}, expected scheme `hdfs://`"), + )); + }; + + let name_node = url.host_str().filter(|h| !h.is_empty()).map(|host| { + url.port() + .map(|port| format!("hdfs://{host}:{port}")) + .unwrap_or_else(|| format!("hdfs://{host}")) + }); + + // `url.path()` borrows from `url` and can't be returned with the input's + // lifetime. Slice the path component out of the original input instead; + // it starts after the first `/` following the `hdfs://` prefix. Opendal + // paths must not start with `/` (`Deleter::delete` rejects them). + let rel = match after_scheme.find('/') { + Some(i) => after_scheme[i..].trim_start_matches('/'), + None => "", + }; + + Ok((name_node, rel)) +} + +/// Creates an operator for the path, cached per effective NameNode (the +/// configured `hdfs.name-node`, else the path authority) — each operator +/// holds an HDFS client with live RPC connections. +pub(crate) fn hdfs_native_create_operator<'a>( + path: &'a str, + config: &HdfsNativeConfig, + operators: &RwLock<HashMap<String, Operator>>, +) -> Result<(Operator, &'a str)> { + let (authority_name_node, relative_path) = hdfs_native_parse_path(path)?; + + let name_node = match config.name_node.clone().or(authority_name_node) { Review Comment: **Major: an empty `hdfs.name-node` silently defeats the path-authority fallback.** `hdfs_native_config_parse` (L35-37) accepts `""` as-is, so `config.name_node` becomes `Some("")` and `.or(authority_name_node)` short-circuits, meaning the authority is never consulted. `init_hdfs_config("")` then filters out the empty entry and emits `dfs.ha.namenodes.nameservice = ""`, producing a confusing failure well downstream instead of the pointed error this function is otherwise careful to give. Worth noting that opendal's own `HdfsNativeBuilder::name_node()` guards this with `if !name_node.is_empty()`, but `Operator::from_config` sets the config struct directly and bypasses the setter, so that guard does not apply here. Suggested: in `hdfs_native_config_parse`, `.filter(|s| !s.trim().is_empty())` before assigning. Trimming a trailing `/` there as well would stop `hdfs://nn:8020/` and a `hdfs://nn:8020` path authority from producing two cache entries for the same cluster. ########## crates/storage/opendal/src/hdfs_native.rs: ########## @@ -0,0 +1,350 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! HDFS storage backend via OpenDAL's `services-hdfs-native` (pure Rust, no JNI). + +use std::collections::HashMap; +use std::sync::RwLock; + +use iceberg::io::{HDFS_HADOOP_CONF_PREFIX, HDFS_NAME_NODE}; +use iceberg::{Error, ErrorKind, Result}; +use opendal::Operator; +use opendal::services::HdfsNativeConfig; +use url::Url; + +use crate::utils::from_opendal_error; + +/// Parse iceberg properties to [`HdfsNativeConfig`]. +pub(crate) fn hdfs_native_config_parse(mut m: HashMap<String, String>) -> Result<HdfsNativeConfig> { + let mut cfg = HdfsNativeConfig::default(); + + if let Some(name_node) = m.remove(HDFS_NAME_NODE) { + cfg.name_node = Some(name_node); + } + + let options: HashMap<String, String> = m + .into_iter() + .filter_map(|(key, value)| { + key.strip_prefix(HDFS_HADOOP_CONF_PREFIX) + .map(|stripped| (stripped.to_string(), value)) + }) + .collect(); + if !options.is_empty() { + cfg.options = Some(options); + } + + Ok(cfg) +} + +/// Parse an HDFS path into `Some("hdfs://<authority>")` (`None` when +/// authority-less) and the relative path (no leading `/`, opendal style). +pub(crate) fn hdfs_native_parse_path(path: &str) -> Result<(Option<String>, &str)> { + let url = Url::parse(path).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}: {e}"), + ) + })?; + // Non-special schemes parse even without `//` (e.g. `hdfs:x` is a valid + // non-hierarchical URL), so require the literal prefix before slicing. + let (Some(after_scheme), "hdfs") = (path.strip_prefix("hdfs://"), url.scheme()) else { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Invalid hdfs path: {path}, expected scheme `hdfs://`"), + )); + }; + + let name_node = url.host_str().filter(|h| !h.is_empty()).map(|host| { + url.port() + .map(|port| format!("hdfs://{host}:{port}")) + .unwrap_or_else(|| format!("hdfs://{host}")) + }); + + // `url.path()` borrows from `url` and can't be returned with the input's + // lifetime. Slice the path component out of the original input instead; + // it starts after the first `/` following the `hdfs://` prefix. Opendal + // paths must not start with `/` (`Deleter::delete` rejects them). + let rel = match after_scheme.find('/') { + Some(i) => after_scheme[i..].trim_start_matches('/'), + None => "", + }; + + Ok((name_node, rel)) +} + +/// Creates an operator for the path, cached per effective NameNode (the +/// configured `hdfs.name-node`, else the path authority) — each operator +/// holds an HDFS client with live RPC connections. +pub(crate) fn hdfs_native_create_operator<'a>( + path: &'a str, + config: &HdfsNativeConfig, + operators: &RwLock<HashMap<String, Operator>>, +) -> Result<(Operator, &'a str)> { + let (authority_name_node, relative_path) = hdfs_native_parse_path(path)?; + + let name_node = match config.name_node.clone().or(authority_name_node) { + Some(name_node) => name_node, + None => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid hdfs path: {path}, authority-less paths require the `{HDFS_NAME_NODE}` property" + ), + )); + } + }; + + // Fast path: check read lock first. + { + let cache = operators + .read() + .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; + if let Some(op) = cache.get(&name_node) { + return Ok((op.clone(), relative_path)); + } + } + + // Slow path: build and insert under write lock, re-checking for a + // concurrent insert. + let mut cache = operators + .write() + .map_err(|_| Error::new(ErrorKind::Unexpected, "HDFS operator cache lock poisoned"))?; + let op = match cache.get(&name_node) { + Some(op) => op.clone(), + None => { + let op = hdfs_native_operator_build(config, &name_node)?; + cache.insert(name_node, op.clone()); + op + } + }; + + Ok((op, relative_path)) +} + +/// Returns the `delete_stream` grouping key for a path: the effective +/// NameNode, mirroring the operator-cache key so paths that resolve to +/// different operators never share a deleter. +pub(crate) fn hdfs_native_batch_key(config: &HdfsNativeConfig, path: &str) -> String { Review Comment: **Major: the NameNode-precedence rule is implemented twice.** `hdfs_native_batch_key` here and `hdfs_native_create_operator` (L99) each independently compute "configured `name_node`, else path authority". These two have to agree: `delete_stream` uses `batch_key_for_path` to pick the deleter and `create_operator` to build it, so if the rules ever drift, paths get grouped onto a deleter built against a different NameNode and deletes silently go to the wrong cluster. Suggested: extract a single `fn hdfs_native_effective_name_node(config: &HdfsNativeConfig, path: &str) -> Result<String>` and have both call it. That also makes `test_hdfs_native_batch_key_configured_name_node_wins` and `test_hdfs_native_create_operator_configured_name_node_wins` redundant, since they currently assert the same rule twice. Separately, `unwrap_or_default()` maps both "unparsable path" and "authority-less with no property" to `""`, so those share a grouping key. Harmless today because `create_operator` then errors on the same path, but it is a silent coupling worth not relying on. ########## crates/storage/opendal/src/lib.rs: ########## @@ -243,6 +262,18 @@ pub enum OpenDalStorage { /// GCS configuration. config: Arc<GcsConfig>, }, + /// HDFS storage variant. + /// + /// The NameNode is taken from the `hdfs.name-node` property when set + /// (comma-separated endpoints enable HA failover), else the path authority. + #[cfg(feature = "opendal-hdfs-native")] + HdfsNative { + /// HDFS configuration. + config: Arc<HdfsNativeConfig>, + /// Operator cache keyed by effective NameNode. + #[serde(skip, default)] + operators: Arc<RwLock<HashMap<String, Operator>>>, Review Comment: **Major: the operator cache becomes part of the public API.** `public-api.txt` gains: ``` pub iceberg_storage_opendal::OpenDalStorage::HdfsNative::operators: alloc::sync::Arc<std::sync::poison::rwlock::RwLock<std::collections::hash::map::HashMap<alloc::string::String, opendal_core::types::operator::operator::Operator>>> ``` No other variant exposes internal state; they all carry only `config`. This pins the cache representation (including the choice of `std::sync::RwLock`) as a semver-relevant detail, so swapping in a different cache later becomes a breaking change. Suggested: wrap it in a newtype with private fields, e.g. `pub struct OperatorCache(RwLock<HashMap<String, Operator>>)`, so only an opaque type appears in the public API. ########## crates/storage/opendal/tests/file_io_hdfs_test.rs: ########## @@ -0,0 +1,280 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Integration tests for HDFS FileIO via OpenDAL `services-hdfs-native`. +//! +//! These tests need the `hdfs-namenode`/`hdfs-datanode` services from +//! `dev/docker-compose.yaml` (started by `make docker-up`); the fixture +//! uses host networking, which needs Linux or a recent Docker runtime. + +#[cfg(feature = "opendal-hdfs-native")] Review Comment: **Blocker: these tests are not `#[ignore]`d, and there is no CI opt-in.** The PR description says "Tests are `#[ignore]`d (host networking is Linux-only); CI opts in via `cargo nextest --run-ignored=only -E 'test(file_io_hdfs)'`". Neither is on the branch: - no `#[ignore]` on any of the 12 tests in this file; - `git diff main...HEAD -- .github/ Makefile` is empty, so there is no workflow step and no `--run-ignored=only`; - `dev/docker-compose.yaml` adds `hdfs-namenode`/`hdfs-datanode` with no `profiles:` key, so `make docker-up` starts them unconditionally. Since the `Makefile` has `test: docker-up` and `ci.yml` runs `--all-targets --all-features`, `opendal-hdfs-native` is enabled and all 12 tests run by default. On macOS/Windows, `network_mode: "host"` does not put the NameNode on the host loopback, so `make test` will fail for every contributor on those platforms. Either add `#[ignore]` plus a Linux-only CI step as described, or put the two compose services behind a `profiles:` key so they are opt-in. While here, the description's test counts look stale too: 16 unit tests in `hdfs_native.rs`, 3 in `lib.rs`, 2 in `resolving.rs` (not 24), and `crates/iceberg/src/io/storage/config/hdfs.rs` has no tests (not 3). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
