This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new ca21f1587 feat(storage): Add HDFS support via opendal
services-hdfs-native (#3111)
ca21f1587 is described below
commit ca21f1587c6caaeb88dd3a9e1cc245747e1aba08
Author: Michael Taranov <[email protected]>
AuthorDate: Thu Oct 22 15:07:39 2026 +0000
feat(storage): Add HDFS support via opendal services-hdfs-native (#3111)
* feat(storage): Add HDFS support via opendal services-hdfs-native
Adds an opendal-hdfs-native cargo feature plus OpenDalStorageFactory::Hdfs
and OpenDalStorage::Hdfs variants in iceberg-storage-opendal, using
OpenDAL's services-hdfs-native (pure-Rust HDFS RPC, no JNI/libhdfs).
The NameNode for a path resolves as: the hdfs.name-node property when set
(comma-separated endpoints enable HA failover), otherwise the path
authority. hadoop.-prefixed properties are forwarded to the HDFS client
configuration, overriding values loaded from $HADOOP_CONF_DIR. Operators
are cached per effective NameNode since each holds live RPC connections.
Revives and updates PR #2441 (by @jordepic) against the current storage
layer and opendal 0.58, where name_node became mandatory and the comma
list is the HA mechanism.
Closes #2440
Co-authored-by: Jordan Epstein <[email protected]>
Co-Authored-By: Claude Opus 5 <[email protected]>
* refactor: address review โ Properties derive for HdfsConfig, drop HDFS CI
special-casing
- HdfsConfig is now pub(crate) and parsed via #[derive(Properties)]
(key/prefix attributes) instead of hand-written TryFrom + TypedBuilder.
- HDFS integration tests are no longer #[ignore]d and the dedicated CI
step is gone; they run under the default nextest invocation since
make docker-up already starts the fixture and the Tests job is
Linux-only.
Co-Authored-By: Claude Opus 5 <[email protected]>
* fix: reject non-hierarchical hdfs URLs; key delete_stream batches by
effective NameNode
- Url::parse accepts non-hierarchical forms like `hdfs:x`; the byte-7
slice then panicked. Require the literal `hdfs://` prefix instead.
- batch_key_for_path grouped by URL host only, so NameNodes differing
by port shared one deleter; key by the effective NameNode (configured
hdfs.name-node, else authority incl. port), matching the operator cache.
Co-Authored-By: Claude Opus 5 <[email protected]>
* refactor(storage): rename hdfs module to hdfs_native, drop unused
HdfsConfig
Per review: the opendal module, its functions and the storage/factory
variants are now hdfs_native-prefixed (leaving room for a libhdfs-backed
variant, see #1130), and the never-consumed HdfsConfig struct is removed
from the core crate โ config/hdfs.rs keeps only the property constants
that iceberg-storage-opendal uses.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* ci: retrigger after setup-python manifest timeout on macos runner
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): address review โ cache newtype, build outside lock, single
NameNode rule, env-gated HDFS tests
- HdfsNativeOperatorCache newtype keeps the cache representation out of
the public API.
- Operators are built outside the cache lock (the build reads Hadoop XML
synchronously); a racing first caller's duplicate is dropped unopened.
- hdfs_native_effective_name_node is the single source of the
configured-else-authority rule for both create_operator and the
delete_stream batch key; unresolvable paths key on themselves.
- hdfs.name-node is trimmed and an empty value is treated as unset, so it
no longer shadows the path-authority fallback.
- HDFS integration tests self-skip unless ICEBERG_TEST_HDFS_ENDPOINT is
set (as the HF tests do) and the compose services sit behind the hdfs
profile; the Linux CI job opts in via env.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* chore(deps): bump rustls to 0.23.45 (RUSTSEC-2026-0285)
Inherited from main (tracked in #3222); same bump as #3165.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* test(storage): pin IPv6 authority handling in HDFS path parsing
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): resolve the HDFS NameNode in relativize_path too
An authority-less path now errors in relativize_path exactly as in
create_operator unless hdfs.name-node is set, so the two cannot disagree.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* docs(storage): state the cost of the synchronous HDFS operator build
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* docs(storage): describe the HDFS operator cache lifecycle
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): normalize each entry of the hdfs.name-node list
A space after a comma was passed through to opendal unchanged, which
silently broke failover to that NameNode.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* feat(storage): accept PyIceberg's hdfs.host/hdfs.port for authority-less
HDFS paths
They map to Hadoop's fs.defaultFS and apply only when the path has no
authority, as in PyIceberg; hdfs.name-node stays the explicit override.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): require a port in every hdfs.name-node entry
An entry without a port can only fail at the first I/O as 'invalid socket
address'; reject it at build time with the property and entry named. Bare
host:port stays accepted, as in Hadoop and opendal.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* docs(storage): state the runtime binding of cached HDFS operators
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* test(storage): cover two-NameNode delete_stream batches and
first-insert-wins caching
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* test(utils): name the HDFS port constant like its siblings
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): rebuild cached HDFS operators once their runtime is gone
Each cache entry parks a sentinel task on the runtime that built the
operator and keeps a Weak to a token the task owns. When that runtime shuts
down the token dies, the entry reads as stale and the operator is rebuilt on
the current runtime instead of hdfs-native panicking on the dead one. Sentinels
are aborted when entries drop.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): resolve HDFS NameNodes like Hadoop does
An authority with a port is used as is; hdfs.name-node serves logical
nameservice authorities and authority-less paths, which otherwise fall back to
fs.defaultFS. Portless strings from any source are logical names that only
hdfs.name-node resolves, reported at resolution instead of at the first I/O.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* refactor(storage): keep the HDFS operator cache out of the public API
OpenDalStorage::HdfsNative now wraps one opaque HdfsNativeStorage with
crate-private fields, built only by the factories. The serialized form keeps
its nesting; the io-timeout propagation test now covers the HDFS factory too.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* perf(storage): build HDFS operators on a blocking thread
The build reads the Hadoop XML config synchronously; create_operator is now
async and offloads it with spawn_blocking, outside the cache lock as before.
The builder itself stays sync and runtime-free.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* ci: run the HDFS integration tests in their own job
The default suite no longer pulls the Hadoop image or waits on the HDFS
health checks; a dedicated required job starts only the two HDFS containers and
runs the HDFS test binary.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): stop dropping HDFS properties silently
PyIceberg's hdfs.user and hdfs.kerberos_ticket log a warning instead of
vanishing (opendal's config has no user field), hdfs.port is validated whether
or not a host is set and warns when it has no host to apply to, and a bare
hadoop. key is rejected.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* test(storage): derive the second NameNode spelling from the endpoint host
Resolving the host to its IPv4 address gives any hostname endpoint a second
batch key; an IP-literal endpoint skips the test explicitly instead of passing
as a no-op.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* chore: record tracing as a dependency of iceberg-storage-opendal in
Cargo.lock
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): report a missing tokio runtime instead of panicking in HDFS
create_operator
The runtime handle is resolved up front and passed into the cache, so the
sentinel is no longer optional and the unreachable no-runtime branch is gone.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): normalize every HDFS NameNode spelling to hdfs://host:port
One parser classifies configured entries, path authorities and
fs.defaultFS: other schemes, logical names, port 0, userinfo, paths and
unbracketed IPv6 are rejected, and bare and prefixed spellings share one cache
key.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* feat(storage): declare HDFS nameservices with hdfs.name-node.<nameservice>
The key expands to Hadoop's dfs.ha.namenodes and dfs.namenode.rpc-address
keys in the forwarded options and the resolver reads them back, so nameservices
declared through hadoop.* work the same way. Plain hdfs.name-node stays the
documented single-cluster default; an undeclared name without it is a pointed
error.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): require a declaration for every logical HDFS nameservice
Plain hdfs.name-node serves authority-less paths only; an undeclared
logical authority is an error naming the key to set instead of a guess that
could reach the wrong cluster.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): reject userinfo and port 0 in HDFS path authorities
Userinfo was silently dropped and port 0 was misread as a logical
nameservice; both are DataInvalid now, and the userinfo message names the host
only so a password never reaches a log.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): validate the NameNode composed from hdfs.host and hdfs.port
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* test(storage): use an authority-less path in the cross-runtime HDFS test
A logical authority now requires a declaration, so the test must take the
plain-key route.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* test(storage): tighten the HDFS tests after review
Drop three redundant unit tests, assert the warnings and the password
redaction directly, make the serde round trip populate the cache first, check
which error each negative case produces, cover the plain key and hdfs.host
serving authority-less paths against the fixture, and name the
declared-nameservice test for what it does.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* fix(storage): mask userinfo in HDFS errors that echo a path or NameNode
A path that failed to parse, or had another scheme, was echoed whole, so
`hdfs://user:pw@nn:bad/x` put the password in the error. The NameNode
config errors (`hdfs.name-node`, `hdfs.name-node.<nameservice>`,
`hdfs.host`/`hdfs.port`, `fs.defaultFS`) echoed their values the same way.
Every `Invalid hdfs path` error now goes through one helper, and the config
errors use the same masking: everything between the scheme and the last
`@` becomes `***`, so a password holding a raw `/`, `?`, `#` or `://` is
covered too.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* fix(storage): recover a poisoned HDFS operator cache lock
A panic elsewhere while the cache lock was held turned every later HDFS
call into a permanent `Unexpected` error. Writes only ever replace whole
entries, so the map stays consistent: both lock sites now recover the
guard with `PoisonError::into_inner`, and `get`/`insert` are infallible.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* test(storage): flush the fake NameNode before sampling dials
The runtime test sampled its dial counter before the listener thread had
necessarily counted every dial from the first runtime. It now drops that
runtime and waits for the listener to close a marker connection; accepts
are served in order, so every earlier dial is counted first. A failed
accept no longer stops the listener.
The runtime-shutdown unit test also asserts that the rebuilt operator is
a new instance, not the stale one with a fresh sentinel.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* docs(storage): state where portless HDFS authorities differ from Hadoop
Hadoop resolves a portless authority from hdfs-site.xml, else dials port
8020. Here it must be declared with `hdfs.name-node.<nameservice>`,
because opendal builds the client as `hdfs://nameservice` and never hands
hdfs-native the authority. The `HDFS_NAME_NODE` and resolver docs no
longer claim Hadoop parity, the error says to add a port or declare the
nameservice, and the README notes the limitation in its own note.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* refactor(storage): rename the HDFS config module to hdfs_native
Matches its siblings, which are named after their storage feature
(azdls, gcs, hf, oss, s3). The module is private and re-exported, so the
public iceberg::io::HDFS_* paths are unchanged.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* test(storage): move the HDFS tests out of lib.rs into hdfs_native
The serde round-trip and relativize_path tests now live next to the code
they cover, with the module's test_hdfs_native_ prefix.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* refactor(storage): read HDFS properties through #[derive(Properties)]
hdfs.name-node, the hdfs.name-node.<nameservice> and hadoop. prefixes,
hdfs.host and hdfs.port are declared in one struct parsed by the
Properties macro instead of being looked up by hand. Host and port stay
trimmed strings so a blank value still counts as unset, as in PyIceberg.
Validation, warnings and error messages are unchanged.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* ci: run the HDFS tests in the default suite like the other fixtures
The HDFS fixture joins the shared bridge network with published ports
instead of using host networking behind a compose profile, so make
docker-up starts it on every platform: the DataNode advertises 127.0.0.1
and the tests set dfs.client.use.datanode.hostname. The tests no longer
skip without ICEBERG_TEST_HDFS_ENDPOINT, and the dedicated Tests (hdfs)
job is gone.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* fix(storage): build HDFS clients in a crate-owned runtime
An hdfs-native client spawns its NameNode connections on the runtime
current when it is built and panics once that runtime is dropped. The
cache used to detect that with a sentinel task per entry, which needed
Handle::try_current, and the build ran under spawn_blocking, which made
create_operator async for every backend. Clients are now built inside a
crate-owned runtime started on first use, so a cached operator works
from any caller's runtime: the sentinel, Handle::try_current and
spawn_blocking are gone, and create_operator is sync again as on main.
The Hadoop XML read (~0.1 ms) runs inline.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* refactor(storage): type HDFS path authorities and validate declarations
at parse
hdfs_native_parse_path returns an HdfsNativeAuthority (NameNode,
Nameservice, Default), so the resolver matches on it instead of
re-parsing the authority to learn whether it has a port. Every HA
nameservice declaration, from hdfs.name-node.<nameservice> or passed
through as hadoop.dfs.ha.namenodes.*, and the fs.defaultFS that serves
authority-less paths are now validated when the config is parsed, so a
broken one fails the first call rather than only the paths that use it.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* refactor(storage): drop index slicing and combinator chains in the HDFS
backend
The nameservice lookup is a for loop with let-else, and the userinfo
masking and the relative-path extraction use split_once/rsplit_once
instead of slicing at computed indices. A test pins that doubled slashes
after the authority are still trimmed.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* fix(storage): attach the hdfs.port parse error as the source
The parse error was interpolated into the message; it is now attached
as the error's source, which iceberg prints after the message.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* fix(storage): share one HDFS client across host name spellings
host_str keeps the host's case for hdfs:// URLs, so hdfs://NN1:8020 and
hdfs://nn1:8020 cached two clients for one NameNode. Endpoint hosts are
now lowercased by one helper used by the config normalizer and the path
parser, as Hadoop's FileSystem cache key does; nameservice names stay
case-sensitive like the Hadoop keys that declare them. The cache's doc
states that it is unbounded by design.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* docs(storage): note the upstream gap behind the HDFS client runtime
opendal's hdfs-native service does not expose hdfs-native's
ClientBuilder::with_io_runtime (Kimahriman/hdfs-native#255); a TODO at
the crate-owned runtime records that a runtime could be passed in once
it does.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* test(storage): make the HDFS HA test fail over for real
The declared-nameservice test listed the single-node fixture twice,
which drove opendal's list path but never needed a failover. Its first
NameNode is now a local port nothing listens on, so every call has to
fail over through opendal's synthetic nameservice; an opendal change
that kept only the first entry of the list would fail it.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* chore: retrigger CI
Docker Hub rate-limited the previous run's image pulls, so the default test
job never started its fixtures.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* chore: retrigger CI
The previous run hit Docker Hub's unauthenticated pull rate limit again
before any test ran.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* docs(storage): document how to enable and configure the HDFS backend
The iceberg-storage-opendal README gains an HDFS section: enabling the
opendal-hdfs-native feature, the factory, the hdfs.* and hadoop.*
properties, the Hadoop config, user and Kerberos it reads, and the
portless-authority limitation. The experimental-backends note now points
to it.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* docs(storage): drop the HDFS README section and link follow-up issues
Configuration docs for every backend are tracked in #3510; the README
keeps the one-line note on the portless-authority limitation. The TODO on
the crate-owned HDFS runtime now points at #3509.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Jordan Epstein <[email protected]>
Co-authored-by: Claude Opus 5 <[email protected]>
---
Cargo.lock | 320 ++++-
crates/iceberg/public-api.txt | 4 +
.../iceberg/src/io/storage/config/hdfs_native.rs | 45 +
crates/iceberg/src/io/storage/config/mod.rs | 2 +
crates/storage/opendal/Cargo.toml | 7 +
crates/storage/opendal/README.md | 3 +
crates/storage/opendal/public-api.txt | 11 +
crates/storage/opendal/src/hdfs_native.rs | 1355 ++++++++++++++++++++
crates/storage/opendal/src/lib.rs | 42 +
crates/storage/opendal/src/resolving.rs | 37 +
crates/storage/opendal/tests/file_io_hdfs_test.rs | 400 ++++++
.../opendal/tests/hdfs_native_runtime_test.rs | 91 ++
crates/test_utils/src/lib.rs | 9 +
dev/docker-compose.yaml | 45 +
dev/hdfs/core-site.xml | 27 +
dev/hdfs/hdfs-site.xml | 42 +
16 files changed, 2428 insertions(+), 12 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index f0463d1a6..a2d7370a0 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",
]
@@ -1137,6 +1148,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"
@@ -1270,7 +1290,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]]
@@ -1345,7 +1374,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]]
@@ -1516,6 +1556,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"
@@ -1680,7 +1726,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]]
@@ -1857,6 +1912,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"
@@ -1934,6 +1998,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"
@@ -2259,6 +2335,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"
@@ -2435,6 +2539,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"
@@ -2956,6 +3101,7 @@ dependencies = [
"serde_json",
"tempfile",
"tokio",
+ "tracing",
"typetag",
"url",
]
@@ -3107,10 +3253,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"
@@ -3369,6 +3525,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"
@@ -3939,6 +4105,7 @@ dependencies = [
"opendal-service-azdls",
"opendal-service-fs",
"opendal-service-gcs",
+ "opendal-service-hdfs-native",
"opendal-service-hf",
"opendal-service-oss",
"opendal-service-s3",
@@ -4093,6 +4260,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"
@@ -4421,8 +4602,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",
@@ -4582,6 +4763,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"
@@ -5232,6 +5445,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"
@@ -5423,7 +5645,7 @@ version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213"
dependencies = [
- "cipher",
+ "cipher 0.4.4",
]
[[package]]
@@ -7286,6 +7508,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"
@@ -7319,13 +7550,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"
@@ -7347,6 +7595,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"
@@ -7359,6 +7613,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"
@@ -7371,12 +7631,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"
@@ -7389,6 +7661,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"
@@ -7401,6 +7679,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"
@@ -7413,6 +7697,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"
@@ -7425,6 +7715,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 87fd3943c..5f87d9881 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -1095,6 +1095,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 000000000..ef9e16bf6
--- /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.<nameservice>` takes the same list and is
+/// sugar for Hadoop's `dfs.ha.namenodes.<nameservice>` and
+/// `dfs.namenode.rpc-address.<nameservice>.<id>`, 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 d8d356de1..8ea4af080 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 d3403598e..83541e89c 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 5db4293e6..ac3a8b62e 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`. A
portless path authority such as `hdfs://nameservice1/warehouse` must be
declared in properties, e.g. `hdfs.name-node.nameservice1 = nn1:8020,nn2:8020`:
unlike Hadoop, nameservices defined only in `$HADOOP_CONF_DIR` are not
resolved, and a portless host is not dialed on the default port 8020.
+
## Usage
Add the crate to your `Cargo.toml` with the feature flags for the backends you
need:
diff --git a/crates/storage/opendal/public-api.txt
b/crates/storage/opendal/public-api.txt
index c2e310335..797d90d1e 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<ope
pub iceberg_storage_opendal::OpenDalStorage::Gcs
pub iceberg_storage_opendal::OpenDalStorage::Gcs::client_config:
iceberg_storage_opendal::OpenDalClientConfig
pub iceberg_storage_opendal::OpenDalStorage::Gcs::config:
alloc::sync::Arc<opendal_service_gcs::config::GcsConfig>
+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<opendal_service_hf::config::HfConfig>
@@ -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<Self, <__D as
serde_core::de::Deserializer>::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<alloc::string::String,
alloc::string::String>) -> iceberg::error::Result<Self>
diff --git a/crates/storage/opendal/src/hdfs_native.rs
b/crates/storage/opendal/src/hdfs_native.rs
new file mode 100644
index 000000000..ef54d8642
--- /dev/null
+++ b/crates/storage/opendal/src/hdfs_native.rs
@@ -0,0 +1,1355 @@
+// 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(#3509): build clients on the catalog's runtime once storage factories
+// receive it, and drop this one.
+static HDFS_NATIVE_RUNTIME: LazyLock<Runtime> = 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<Operator> {
+ 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 9e0155201..693f4c18b 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 51e76da47..d7a977e7a 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 000000000..ef9da20a1
--- /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<dyn StorageFactory>) -> 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<String> {
+ let url = url::Url::parse(endpoint).ok()?;
+ let host = url.host_str()?;
+ if host.parse::<std::net::IpAddr>().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.<nameservice>` 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<String> = (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 000000000..da22260d7
--- /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<AtomicUsize>) {
+ 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 bf2dad0ed..176767813 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<C: Catalog>(catalog: &C, ns:
&NamespaceIdent) {
diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml
index cb89e3042..3bf9d730b 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 000000000..273676bba
--- /dev/null
+++ b/dev/hdfs/core-site.xml
@@ -0,0 +1,27 @@
+<!--
+ 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.
+-->
+
+<configuration>
+ <property>
+ <name>fs.defaultFS</name>
+ <value>hdfs://hdfs-namenode:8020</value>
+ </property>
+ <property>
+ <name>hadoop.http.staticuser.user</name>
+ <value>root</value>
+ </property>
+</configuration>
diff --git a/dev/hdfs/hdfs-site.xml b/dev/hdfs/hdfs-site.xml
new file mode 100644
index 000000000..b9bf0c63c
--- /dev/null
+++ b/dev/hdfs/hdfs-site.xml
@@ -0,0 +1,42 @@
+<!--
+ 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.
+-->
+
+<configuration>
+ <property>
+ <name>dfs.replication</name>
+ <value>1</value>
+ </property>
+ <property>
+ <name>dfs.permissions.enabled</name>
+ <value>false</value>
+ </property>
+ <property>
+ <name>dfs.namenode.name.dir</name>
+ <value>file:///tmp/hadoop-root/dfs/name</value>
+ </property>
+ <property>
+ <name>dfs.datanode.data.dir</name>
+ <value>file:///tmp/hadoop-root/dfs/data</value>
+ </property>
+ <!-- What clients outside the container network dial for the DataNode,
+ through its published port, when they set
+ dfs.client.use.datanode.hostname. -->
+ <property>
+ <name>dfs.datanode.hostname</name>
+ <value>127.0.0.1</value>
+ </property>
+</configuration>
