mixermt commented on code in PR #3111: URL: https://github.com/apache/iceberg-rust/pull/3111#discussion_r4035433608
########## 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: You're right — that paragraph was left over from before the `#[ignore]` step was removed at @blackmwk's request, and the counts were stale. Fixed in d17ae38 by following the HF tests' approach: the HDFS tests skip unless `ICEBERG_TEST_HDFS_ENDPOINT` is set, the compose services sit behind a `hdfs` profile, and the Linux CI job opts in via env. `make test` now skips them on any platform. Description updated. ########## 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: Agreed, fixed in d17ae38: the operator is built outside the lock and inserted with `entry().or_insert`, so a racing first caller may build a duplicate that gets dropped before any I/O. One note on scope: `NameServiceProxy::new` doesn't connect (connections are lazy on first RPC), so the work under the old lock was two local XML reads per NameNode — still not worth holding a write lock for. ########## 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: Fair point. d17ae38 wraps the cache in `HdfsNativeOperatorCache` (private field, `Default`), so the public API only names that type and the internal representation can change freely. ########## 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: Done in d17ae38: both `create_operator` and the batch key go through one `hdfs_native_effective_name_node`. Unresolvable paths now key on the full path (same as `hf_batch_key`) instead of `""`. Dropped the duplicate precedence test. ########## 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: Good catch — `from_config` does bypass the builder's empty-string guard. d17ae38 trims the property, strips a trailing `/`, and treats an empty value as unset, with tests for both cases. ########## dev/docker-compose.yaml: ########## @@ -147,6 +147,49 @@ services: timeout: 5s retries: 5 + # ============================================================================= + # HDFS - single-node NameNode + DataNode for HDFS tests + # ============================================================================= + # hdfs-native connects to DataNodes by their registered IP — unroutable on + # a docker bridge, hence host networking (needs Linux or a recent runtime). + hdfs-namenode: + image: apache/hadoop:3.5.0 + network_mode: "host" + command: ["hdfs", "namenode"] + environment: + ENSURE_NAMENODE_DIR: "/tmp/hadoop-root/dfs/name" + extra_hosts: + - "docker-desktop:127.0.0.1" + volumes: + - ./hdfs/core-site.xml:/opt/hadoop/etc/hadoop/core-site.xml:ro + - ./hdfs/hdfs-site.xml:/opt/hadoop/etc/hadoop/hdfs-site.xml:ro + healthcheck: + test: ["CMD-SHELL", "hdfs dfsadmin -safemode get | grep -q OFF"] + interval: 5s + timeout: 15s + retries: 30 + start_period: 30s + + hdfs-datanode: Review Comment: It's possible: with host networking, each extra DataNode needs its own `dfs.datanode.{address,http.address,ipc.address}` ports and data dir. I kept the fixture single-node to match opendal's own HDFS fixture — DataNode count doesn't change the FileIO adapter's behavior (failover and erasure-coded reads are hdfs-native's domain), and it's already the slowest service in `make docker-up`. Happy to add a two-DataNode variant as a follow-up if there's a scenario you'd like covered. -- 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]
