blackmwk commented on code in PR #3111:
URL: https://github.com/apache/iceberg-rust/pull/3111#discussion_r4346896261
##########
crates/storage/opendal/README.md:
##########
@@ -82,3 +85,40 @@ async fn main() -> iceberg::Result<()> {
Ok(())
}
```
+
+## HDFS
Review Comment:
Why we need to add this section for hdfs only? I'm not against it, but I
think we should add docs for all storages. Please remove this section and
create an issue as follow up to add them.
##########
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.<nameservice>`
+/// 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<String> {
+ 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<Vec<String>> {
+ 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.<nameservice>`, 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<String>,
+ /// `hdfs.name-node.<nameservice>` declarations, keyed by nameservice.
+ #[property(prefix = HDFS_NAME_NODE_DECLARATION_PREFIX)]
+ nameservices: HashMap<String, String>,
+ #[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<String, String>,
+}
+
+// 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<Vec<String>> {
+ hdfs_native_name_node_list(HDFS_NAME_NODE, value)
+}
+
+fn parse_trimmed(value: &str) -> Result<String> {
+ 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<String, String>) ->
Result<HdfsNativeConfig> {
+ 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<String, String>,
+ mut warn: impl FnMut(String),
+) -> Result<HdfsNativeConfig> {
+ let props = HdfsNativeProperties::from_properties(&m)?;
+ let mut options = props.hadoop;
+
+ // `hdfs.name-node.<nameservice>` 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<String> = (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::<u16>().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<Option<String>> {
+ 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<Option<String>> {
+ 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<HdfsNativeConfig>,
+ #[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<RwLock<HashMap<String,
Operator>>>);
+
+// 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<Operator> {
+ 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<Runtime> = LazyLock::new(|| {
Review Comment:
Please create an issue to pass catalog runtime to storage factor so that we
could remove this.
--
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]
