feat(storage): Add HDFS support via opendal services-hdfs-native by mixermt · Pull Request #3111 · apache/iceberg-rust

mixermt · 2026-08-30T06:58:22Z

Which issue does this PR close?

Closes Support tables which live on HDFS #2440.
Part of HDFS support in iceberg-rust #1130.

This revives #1131's successor #2441 by @jordepic (closed by the stale bot after a first review round), rebased onto the current Storage-trait layout and updated for opendal 0.58, where services-hdfs-native changed behavior in ways that required design changes (details below).

What changes are included in this PR?

Adds an opendal-hdfs-native cargo feature to iceberg-storage-opendal, with OpenDalStorageFactory::HdfsNative / OpenDalStorage::HdfsNative variants and hdfs:// routing in OpenDalResolvingStorage. The backend uses OpenDAL's services-hdfs-native (pure-Rust HDFS RPC via hdfs-native — no JNI/libhdfs). The feature is experimental and not part of opendal-all.

NameNode resolution (differs from #2441, forced by opendal's hdfs-native service, where name_node is mandatory and, since 0.56 (apache/opendal#7248), its comma-split list is the HA mechanism):

An authority with a port (hdfs://host:port/path) is used as is; a configured hdfs.name-node never overrides it. Userinfo and port 0 in an authority are rejected with DataInvalid (every error the HDFS backend raises for a path or NameNode value masks everything between the scheme and the last @, e.g. hdfs://***@nn:8020/x).
A portless authority (hdfs://ns1/path) must be a declared nameservice: hdfs.name-node.<nameservice> takes a single endpoint or a comma-separated list for HA failover and is sugar for Hadoop's dfs.ha.namenodes.<nameservice> / dfs.namenode.rpc-address.<nameservice>.<id>, which are honored as well when passed through hadoop.*. Unlike Hadoop, an undeclared one is an error naming the key to set, rather than being resolved from $HADOOP_CONF_DIR or dialed on port 8020: opendal always builds the client as hdfs://nameservice from an explicit list, so hdfs-native never sees the path's authority (noted in the README; lifting it needs opendal to accept a logical nameservice as name_node). Every NameNode spelling, configured or from a path, is normalized to hdfs://host:port (host lowercased, as Hadoop's FileSystem cache key does) so all sources share cache keys; other schemes, userinfo, paths, port 0 and unbracketed IPv6 are rejected, and hdfs-native has no default port, so a portless NameNode can only be a logical name.
An authority-less path (hdfs:///path) resolves through plain hdfs.name-node (new HDFS_NAME_NODE / HDFS_HADOOP_CONF_PREFIX constants in iceberg::io), else fs.defaultFS: set from PyIceberg's hdfs.host/hdfs.port (new HDFS_HOST / HDFS_PORT constants, port defaults to 8020, validated at config time; as in PyIceberg they apply only without a path authority) or from hadoop.fs.defaultFS. Malformed declarations, sugar or passed through, and a non-HDFS fs.defaultFS that serves authority-less paths fail when the config is parsed; an undeclared nameservice or a missing default fails when a path needs it, with an error naming the property to set.

The properties are read through #[derive(Properties)]. hadoop.-prefixed properties pass through to the HDFS client config with the prefix stripped, overriding $HADOOP_CONF_DIR values (mirroring the hadoop. catalog-property convention of the Java integrations); hdfs-native still loads core-site.xml/hdfs-site.xml from $HADOOP_CONF_DIR/$HADOOP_HOME for everything else, and Kerberos works via libgssapi_krb5 (runtime dlopen). PyIceberg's hdfs.user/hdfs.kerberos_ticket have no equivalent in opendal's config and log a warning.

Operators are cached per effective NameNode inside an opaque HdfsNativeStorage (built only by the factories). hdfs-native binds a client's NameNode connections to the runtime current when the client is built and panics once that runtime is gone, so clients are built inside a crate-owned runtime (started on first use) and a cached operator works from any caller's runtime; create_operator stays synchronous for every backend. The build reads the Hadoop XML synchronously (~0.1 ms), once per NameNode per storage. A poisoned cache lock is recovered rather than failing every later call.

Relative paths are returned opendal-style without a leading / — opendal::Deleter::delete (used by delete_stream) rejects leading slashes, which an integration test caught.

Test infrastructure: a single-node HDFS docker fixture (apache/hadoop:3.5.0) on the shared bridge network with published ports, started by make docker-up like the other fixtures. The DataNode advertises 127.0.0.1 and the tests set dfs.client.use.datanode.hostname, since hdfs-native otherwise dials the DataNode's container IP, which the host cannot reach on macOS or Windows. The DataNode healthcheck gates on NameNode registration so --wait means writable. The tests run in the default CI job; ICEBERG_TEST_HDFS_ENDPOINT overrides the endpoint.

Are these changes tested?

34 unit tests across config parsing and normalization (per-entry trimming, port validation, hdfs.host/hdfs.port into fs.defaultFS, rejected and ignored keys, declarations validated at parse), path parsing (authority/port/IPv6/authority-less/non-hierarchical), NameNode precedence and nameservice declarations, spelling normalization (including host case), operator caching (first-insert-wins for racing builds, poisoned-lock recovery), userinfo masking in errors, serde round-trip, batch keys, relativize, and scheme resolution.
14 integration tests against the docker fixture covering exists/read/write/overwrite/metadata/range reads/streaming writes/delete/delete_prefix/delete_stream (+empty, +two NameNode spellings), authority-less paths through hdfs.name-node and hdfs.host/hdfs.port, resolving storage over hdfs://, and the HA flow (logical authority in the path, declared with a dead first NameNode so every call fails over), plus a fixture-free test that reuses a FileIO from a second runtime after the first is dropped (it panics if clients are built on the caller's runtime). The docker-backed suite runs in the default CI job.
make check (incl. cargo-deny licenses), full-workspace unit/doc tests, MSRV, the public API check, and the s3/gcs/hms/resolving integration suites pass locally; public-api.txt regenerated for both crates.

AI Disclosure

Developed with AI assistance (Claude Code): drafting code/tests/fixtures starting from #2441, and cross-checking the design against the locked opendal 0.58.1 / hdfs-native 0.14.6 sources. I reviewed the implementation and ran all verification locally. Areas worth reviewer attention: the NameNode-resolution semantics above (opendal's synthetic-nameservice behavior constrains what hdfs://<nameservice> paths can do without the property), and the Windows --all-features build of hdfs-native, which I could only verify via CI.

🤖 Generated with Claude Code

@jordepic

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 apache#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 apache#2440 Co-authored-by: Jordan Epstein <jordepic@users.noreply.github.com> Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

blackmwk

I found two correctness issues that need to be addressed before merge: malformed hdfs: URLs can panic, and bulk deletion can conflate NameNodes running on different ports. Details and suggested regression coverage are inline. Please also resolve the two existing review threads.

This review was drafted by an AI-assisted tool and confirmed by an Apache Iceberg Rust maintainer. After you've addressed the points above and pushed an update, an Apache Iceberg Rust maintainer — a real person — will take the next look at the PR. The findings cite the project's review criteria; if you think one of them is mis-applied, please reply on the PR and a maintainer will weigh in.

More on how Apache Iceberg Rust handles maintainer review: CONTRIBUTING.md.

…S 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 <noreply@anthropic.com>

…eat-storage-hdfs-native

…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 <noreply@anthropic.com>

blackmwk

Thanks @mixermt for this pr. I think the variant should be named to Hdfs-native rather than hdfs. Hdfs should be reserved for non native hdfs support.

# Conflicts: # Cargo.lock

…sConfig 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 apache#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 <noreply@anthropic.com>

mixermt · 2026-09-05T20:20:06Z

@blackmwk
I've addressed all the recent feedback.
Could you please give this one last review? 🙏

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

comphead

Reviewed the backend against the locked opendal-service-hdfs-native 0.58.1 and hdfs-native 0.14.5 sources, and against the existing hf/oss/azdls backends in this crate.

The overall shape fits the crate well: *_config_parse / *_batch_key / (Operator, &str) mirror hf, and the dedicated batch_key arm is genuinely needed, since host-only keying would merge NameNodes that differ only by port. The operator cache is justified too, given each client re-reads the Hadoop XML and opens RPC connections.

Leaving 5 inline comments: 2 blockers (test gating versus what the description claims, and blocking I/O under the cache write lock) and 3 majors (the cache leaking into the public API, duplicated NameNode-precedence logic, and empty-name-node handling).

I also have a handful of minor simplification notes and two questions (operator caching pins the Client to the tokio runtime that first built it, since hdfs-native captures Handle::try_current() eagerly; and the hadoop. prefix scope). Happy to add those if useful, but they seemed like noise next to the above.

…gle 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 <noreply@anthropic.com>

Inherited from main (tracked in apache#3222); same bump as apache#3165. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

mixermt · 2026-09-17T09:55:46Z

Thanks @comphead for the thorough review — all five items are addressed in d17ae38 and answered inline. The security_audit failure was RUSTSEC-2026-0285 in rustls 0.23.41, which is main's current pin (AWS SDK chain, unrelated to hdfs-native; tracked in #3222). Bumped to 0.23.45 in b8cc860, the same fix #3165 applied. The minor simplification notes you held back are welcome — please send them.

"Bumped to 0.23.45 in b8cc860, the same fix #3165 applied" if we keep it without it CI will stay red, tell me if we want to revert this under this PR

# Conflicts: # Cargo.lock

mixermt · 2026-09-22T12:05:23Z

@comphead @blackmwk any other comments?
I really want to push this forward

I'm testing this with datafusion-comet, so far no issues in this code.

mixermt · 2026-10-10T18:51:51Z

@comphead Done: the iceberg-storage-opendal README now has an HDFS section covering how to enable opendal-hdfs-native, the factory, the hdfs.*/hadoop.* properties, the Hadoop config, user and Kerberos it reads, and the portless-authority limitation.

…-rust#3111 ef451b66 merges apache/iceberg-rust main into the PR: typed NULL arrays for schema-added nested columns on read (apache#2635), an azdls relative-path fix (apache#3311) and dependency bumps. The HDFS resolver is unchanged, so the hdfs.name-node.<nameservice> declaration Comet emits still applies, and the head still contains main's pin (af1da4c5). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Brings in origin/main up to 9e13649 and moves the iceberg-rust pin to the current head of apache/iceberg-rust#3111 (0145ca6b), which still contains main's pin (af1da4c5). Main now loads the Iceberg scheme allow-lists from the native builtin_storage_schemes over JNI, so hdfs is admitted there (read and write) instead of in the two Scala sets. The pin brings apache/iceberg-rust apache#3145, which makes FileScanTaskDeleteFile's fields private behind a validating builder and its deletion-vector offsets unsigned: the planner builds delete files through the builder (rejecting a negative offset or size as iceberg-rust does for a manifest entry), and the delete-file sizing rebuilds each delete through it. The reworked HDFS backend validates every declared nameservice and a forwarded fs.defaultFS when it builds the storage, failing all I/O otherwise. Comet's gate now simulates its option expansion (hadoop.* keys win over the hdfs.name-node.<ns> sugar key by key) and declines those cases at planning. It also builds HDFS clients in a runtime of its own, so the AbortOnDrop comment about runtime-bound operators is replaced. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

comphead · 2026-10-12T18:57:37Z

@blackmwk PTAL as you requested changes

laskoviymishka

@blackmwk can you PTAL? This is LGTM from my side.

blackmwk

Thanks @mixermt for this pr, just two extra comments, others LGTM!

Configuration docs for every backend are tracked in apache#3510; the README keeps the one-line note on the portless-authority limitation. The TODO on the crate-owned HDFS runtime now points at apache#3509. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

blackmwk

Thanks @mixermt for this pr!

comphead · 2026-10-22T16:17:03Z

Thanks everyone to work, review and support on this PR, it unblocks a vast number of use cases.

Teamwork 💪

jordepic · 2026-10-22T16:19:56Z

Stupendous 🔥

Cannot wait to use it! May have to speed forward on the iceberg-rust release!

laskoviymishka · 2026-10-22T18:18:02Z

Super team! Thanks @mixermt / @comphead / @blackmwk !

Resolve conflicts with HDFS support (apache#3111): the OpenDAL S3 and GCS features enable the now-optional `tracing` dependency, and the new `HdfsNative` variant is `#[non_exhaustive]` like the others.

Port the vended credential provider to contextual sessions (apache#3081): credential requests are authenticated by a table session derived from the contextual session of the operation that loaded the table, and the table-scoped client no longer carries a session. Resolve conflicts with HDFS support (apache#3111) and ADLS relative paths (apache#3311).

Move each backend's tests out of the crate-level `tests` module into the backend's own module, following the hdfs_native split in apache#3111. `lib.rs` keeps only the cross-backend tests: client config parsing, factory construction and the storage-agnostic writer contract. Feature gates are not repeated on the moved tests: `lib.rs` already guards each backend module behind its `opendal-*` feature in `cfg_if!`, so a test in `gcs.rs` can only be compiled when that module exists. `tests::client_config` becomes `pub(super)` because the relocated timeout test in `memory.rs` calls it. Generated-by: OpenAI Codex 0.165.0-alpha.1.2

Move each backend's tests out of the crate-level `tests` module into the module that owns the backend, following the `hdfs_native` split in apache#3111: `memory.rs`, `fs.rs`, `s3.rs`, `gcs.rs`, `oss.rs` and `azdls.rs` each get their own `mod tests`. `lib.rs` keeps the cross-backend tests: client config parsing, factory construction, the storage-agnostic writer contract, and the IO timeout test, which exercises the `TimeoutLayer` that `resolve` applies to every backend. Feature gates are not repeated on the moved tests: `lib.rs` already guards each backend module behind its `opendal-*` feature in `cfg_if!`, so a test in `gcs.rs` can only be compiled when that module exists. Generated-by: OpenAI Codex 0.165.0-alpha.1.2

## Which issue does this PR close? - Closes issue 3385. ## What changes are included in this PR? Moves each backend's tests out of the crate-level `tests` module into the module that owns the backend, following the `hdfs_native` split in apache#3111: `memory.rs`, `fs.rs`, `s3.rs`, `gcs.rs`, `oss.rs` and `azdls.rs` each get their own `mod tests`. `lib.rs` keeps the cross-backend tests: client config parsing, factory construction, the storage-agnostic writer contract, and the IO timeout test, which exercises the `TimeoutLayer` that `resolve` applies to every backend. Feature gates are not repeated on the moved tests. `lib.rs` already guards each backend module behind its `opendal-*` feature in `cfg_if!`, so a test in `gcs.rs` can only compile when that module exists. No test was added, removed or renamed; the moved bodies are unchanged. ## Are these changes tested? `cargo fmt --all -- --check`, `cargo clippy -p iceberg-storage-opendal --all-targets -- -D warnings` and the same with `--all-features`, plus `cargo test -p iceberg-storage-opendal --lib` (17 tests) and with `--all-features` (74 tests). ## AI Disclosure Codex and Grok Bot, working for dragonknightk, contributed to this code.

## Which issue does this PR close? - Closes issue 3385. ## What changes are included in this PR? Moves each backend's tests out of the crate-level `tests` module into the module that owns the backend, following the `hdfs_native` split in apache#3111: `memory.rs`, `fs.rs`, `s3.rs`, `gcs.rs`, `oss.rs` and `azdls.rs` each get their own `mod tests`. `lib.rs` keeps the cross-backend tests: client config parsing, factory construction, the storage-agnostic writer contract, and the IO timeout test, which exercises the `TimeoutLayer` that `resolve` applies to every backend. Feature gates are not repeated on the moved tests. `lib.rs` already guards each backend module behind its `opendal-*` feature in `cfg_if!`, so a test in `gcs.rs` can only compile when that module exists. No test was added, removed or renamed; the moved bodies are unchanged. ## Are these changes tested? Existing unit tests were relocated unchanged. `cargo test -p iceberg-storage-opendal --lib` runs 26 tests with default features and 83 with `--all-features`, matching upstream/main. ## AI Disclosure Human-led contribution. Grok Bot, working for dragonknightk, assisted with this change. All submitted content was reviewed by a human before submission. (๑•̀ㅂ•́)و✧

…-rust's hdfs-native backend (apache#5898) * feat: read and write Iceberg tables on HDFS via iceberg-rust's hdfs-native backend Adds `hdfs://` to the native Iceberg scan and write paths, backed by iceberg-rust's pure-Rust `hdfs-native` OpenDAL backend (apache/iceberg-rust#3111). This is a second, independent HDFS client: the plain-Parquet native scan reaches HDFS through libhdfs/JNI (`fs.comet.libhdfs.schemes`), while an Iceberg table is opened over pure-Rust RPC. The two share only the `$HADOOP_CONF_DIR` XML and authenticate separately. The scheme arm alone is not sufficient on an HA cluster. opendal's `HdfsNativeBuilder` never dials the authority written in the path: it builds its client against a synthetic authority and synthesizes `dfs.ha.namenodes.<synthetic>` / `dfs.namenode.rpc-address.<synthetic>.nnN` from the comma-separated `hdfs.name-node` value (`init_hdfs_config`). iceberg-rust falls back to the path authority only when that property is absent, which is correct just for a single-NameNode cluster. An HA table location reads `hdfs://<nameservice>/...`, and a nameservice is not a routable host, so without a NameNode list every HA table would fail to connect at execution time -- after the planner had already committed to the native scan. `hadoopToIcebergHdfsProperties` therefore resolves the endpoints from the session Hadoop configuration (`dfs.ha.namenodes.<ns>` plus each `dfs.namenode.rpc-address.<ns>.<nn>`), joined in declaration order, so a standard HDFS client configuration needs no extra settings. An explicit catalog `hdfs.name-node` still wins. A partially resolved list yields nothing rather than a short failover list, which would silently turn a failover into an outage. Because one `hdfs.name-node` overrides the authority of every path the FileIO opens, a scan whose data/delete files span more than one HDFS authority now falls back: the second nameservice would otherwise be read from the first one's NameNode at the same relative path, returning wrong data rather than an error. Changes: - `storage_factory_for`: `hdfs` arm; `hdfs.`/`hadoop.` added to `STORAGE_PROPERTY_PREFIXES` so the NameNode list and client overrides reach FileIO - `CometScanRule.icebergReadableSchemes` and `CometIcebergNativeWrite.SupportedStorageSchemes`: admit `hdfs` - `IcebergTaskValidationResult.dataFileHdfsAuthorities` + multi-authority fallback - `native/Cargo.toml`: enable `opendal-hdfs-native` Testing: - `CometIcebergHdfsSuite`: end-to-end reads against an in-process MiniDFSCluster (plain read, pushdown filter, partitioned table across multiple data files), asserting the data location is genuinely `hdfs://` - 5 unit tests pinning the HA NameNode translation, 2 pinning the native scheme and property forwarding - Full Rust workspace suite (1442 tests) and the three affected JVM suites pass Note: writes are gated and property-forwarded but have no functional test; only reads are covered end to end. The HA translation is unit-tested only, as MiniDFSCluster is single-NameNode. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs: trim duplicated commentary around the HDFS NameNode resolution The opendal synthetic-authority mechanic -- that `hdfs.name-node` overrides the authority of every path the FileIO opens -- was spelled out at seven sites. Keep it once, in `hadoopToIcebergHdfsProperties`'s scaladoc, and leave short pointers at the rest. Also drops a paragraph that narrated the eight lines of code beneath it, a `@param` restating its own signature, and a sentence duplicated four lines apart inside `hadoopToIcebergHdfsProperties`. Retained the reasoning that cannot be recovered from the code: why opendal ignores the path authority, why the NameNode list is all-or-nothing, why `getRawAuthority` rather than `getHost`, and that hdfs-native and libhdfs/JNI are two separate clients. Comments only; no behaviour change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test: skip the Iceberg HDFS suite when MiniDFSCluster cannot start `hadoop-client-minicluster` is pinned at 3.3.4 while Spark supplies `hadoop-client-api`/`runtime` 3.4.2 on the Spark 4.x profiles, so `HttpServer2` resolves a shaded Jetty class the older jar does not carry and the NameNode web server fails to start. Record that in `beforeAll` and cancel the tests instead of aborting the suite, so CI stays green on profiles where the fixture cannot run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: match hadoop-client-minicluster to the hadoop each Spark profile supplies `hadoop.version` was a single global 3.3.4, but only Spark 3.4/3.5 ship hadoop 3.3.x. On the Spark 4.x profiles `HttpServer2` comes from Spark's newer hadoop-client-runtime and resolves shaded Jetty classes the 3.3.4 minicluster does not carry, so MiniDFSCluster's NameNode web server fails to start and any suite using `WithHdfsCluster` cannot run. Set `hadoop.version` per Spark profile, as `parquet.version` already is: 3.4 -> 3.3.4, 3.5 -> 3.3.4, 4.0 -> 3.4.1, 4.1 -> 3.4.2, 4.2 -> 3.5.0. The 3.x profiles are unchanged; a global bump would only have moved the skew onto them. `CometIcebergHdfsSuite` now runs on the default profile instead of cancelling. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: report task input metrics for fused and shuffled native Iceberg scans The scan input bridge only ran when the native block's root was the scan itself. `hasScanInput` recognised `CometNativeScanExec` alone, so an Iceberg scan fused below a filter, project or aggregate, or feeding the native shuffle writer, never reported bytesRead/recordsRead and the Spark UI showed "Input: 0 B / 0 records" for those stages. The listener was also registered after `CometExecIterator`, so it ran before the final native metric flush whenever the iterator was not drained. Register the bridge in `CometExecRDD.compute` ahead of the iterator, drop the per-exec `compute` overrides and the `hasScanInput` gate (the bridge is a no-op without `bytes_scanned` leaves), and increment rather than set the task input metrics, claiming each accumulator once per task through the registry the spill reporter already uses so overlapping or repeated registrations count once while other input sources still add up. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix: register the native shuffle input bridge ahead of nested inputs CometNativeShuffleWriter registered reportScanInputMetrics after CometNativeShuffleInputRDD had resolved its inputs, so for a nested native block that a failed or undrained writer plan left unflushed, the writer's listener claimed the nested scan leaves before their iterator published final values and the nested block's own listener then added nothing. Register scan input reporting next to spill reporting in CometNativeShuffleInputRDD.compute, before the input producers and the writer's iterator, and drop the writer's registration. Rename the RDD's metric node to taskMetricNode since it now feeds both task metrics, and state that a coalesced task's repeated registrations are deduplicated rather than summed. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * feat: show Iceberg planning metrics on the native scan node in the SQL UI The SQL UI aggregates accumulator updates from listener events and never reads driver-side values. Iceberg posts its planning metrics (manifests and files scanned or skipped, planning duration, file sizes) for the original BatchScanExec's accumulators, whose ids are not in the Comet plan graph, and CometIcebergNativeScanExec never posted its own, so every planning metric showed 0. Under DPP the driver-side values were 0 as well, because planning runs on a copy of the original plan while the lazy metrics read the original's accumulators. Post the planning metrics as a driver metric update once per SQL execution from the first execution-time read of commonData, read them from the plan copy that actually planned, register them under Iceberg's descriptions so the labels match BatchScan, and keep them out of the executor metric tree so tasks do not report zeros into the per-task statistics. Document that output rows and recordsRead count rows after iceberg-rust's row filter, and that row deletes applied has no native counterpart. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * style: wrap the resolve-needed fallback reason under the 100-column scalastyle limit The reason for an unresolved nameservice now names the Hadoop configuration the declaration is derived from (session overlaid with the table FileIO's) rather than the session alone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: move the iceberg-rust pin to the latest head of apache/iceberg-rust#3111 ef451b66 merges apache/iceberg-rust main into the PR: typed NULL arrays for schema-added nested columns on read (apache#2635), an azdls relative-path fix (apache#3311) and dependency bumps. The HDFS resolver is unchanged, so the hdfs.name-node.<nameservice> declaration Comet emits still applies, and the head still contains main's pin (af1da4c5). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat(iceberg): read Iceberg tables on HDFS natively via iceberg-rust's hdfs-native backend Moves the iceberg-rust pin to apache/iceberg-rust ca21f158, which carries the HDFS backend (opendal-hdfs-native, a pure-Rust HDFS client), and routes hdfs to it for reads. Writes keep Iceberg's writer. That backend dials an authority with a port as written and resolves a portless one, an HA nameservice, only through a declaration in the FileIO properties, never through $HADOOP_CONF_DIR. The scan rule now collects the HDFS authorities of the data and delete files and declares each portless one as hdfs.name-node.<ns> from the session's dfs.ha.namenodes.<ns> and dfs.namenode.rpc-address.<ns>.<id>, merged before the catalog's properties so a catalog's own declaration wins. A nameservice that stays undeclared, and any HDFS read on a cluster whose hadoop.security.authentication is not simple (the client cannot use an executor's delegation tokens), fall back at planning instead of failing on an executor. hadoop.version, which pins the hadoop-client-minicluster the HDFS test fixture needs, now follows each Spark profile's Hadoop line: with a global 3.3.4, MiniDFSCluster cannot start on the Spark 4.x profiles. CometIcebergHdfsSuite reads tables on a single-NameNode cluster and on a two-NameNode HA cluster addressed by nameservice, before and after a failover and with the first NameNode shut down. CometScanSchemeFallbackSuite drives the real scan rule over Iceberg tables whose manifests name hdfs:// data files. The pin also changes how iceberg-rust matches nested fields in data files that have neither field ids nor a name mapping: a read of only some of a struct's fields now fails rather than guessing by position. The test for such files expects that, and the Iceberg guide documents it. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): fall back for name-mapped tables that read list or map columns iceberg-rust at the pinned revision reads a data file that has no field ids through the table's name mapping, but cannot open the list layouts Hive and Spark's writeLegacyFormat write (a two-level list, or a three-level one with bag/array_element groups) that way: the read schema it supplies keeps a field id on the list's element field, and the Parquet reader rejects it as incompatible. A file's layout is only known from its footer, so a scan of a table with schema.name-mapping.default now falls back whenever it reads a list or map column, at any depth. Iceberg's TestCreateActions migrates such files and reads them back with Comet enabled; its Hive-style and two-level list cases failed without this. CometIcebergNativeSuite imports a legacy-format file under a name mapping and checks that the list column falls back while the primitives stay native. The limitation is listed in the Iceberg guide. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): decline scans that mix storage backends or read HDFS encryption zones The native scan builds one FileIO from the metadata location and opens every data and delete file through it, so a file on another backend (hdfs:// data under local metadata, a local file under HDFS metadata, S3 data under local metadata) failed at open time. CometScanRule now declines a scan whose data/delete files are not all in the metadata location's storage family (file, s3/s3a, each S3-compliant alias, gs, oss, hdfs). hdfs-native is built without its kms feature and rejects every file of an HDFS encryption zone. When the NameNode advertises a key provider or the session configures one, the rule checks each distinct data/delete directory with FileStatus.isEncrypted() and falls back. The planner tests keep their warehouse on the fake hdfs:// filesystem so metadata and data share a storage; CometIcebergHdfsSuite gains a cluster with a jceks key provider and an encryption zone. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): claim HDFS scans only as the Spark user, batch the encryption-zone probe, default the NameNode port The native client connects as HADOOP_USER_NAME or the executor's OS user, never as the Spark user the JVM reader runs as, and OpenDAL exposes no user setting. The planner now claims an HDFS scan only when the executors are known to read as the Spark user (spark.executorEnv.HADOOP_USER_NAME set to it; in local mode the JVM's user) and falls back naming the setting otherwise. The encryption-zone probe groups data directories by parent and lists each parent once (the NameNode returns every child's encrypted flag), caching the flags per filesystem and directory, so a partitioned table costs one call instead of one per partition. An undeclared portless authority is a single NameNode on Hadoop's default port, 8020, as DFSUtilClient.getNNAddress dials it for the JVM client; a name in dfs.nameservices without NameNodes still falls back, and a catalog's own hadoop.dfs.ha.namenodes.<ns> keeps precedence. Tests: an end-to-end scan over data files on two NameNodes, the encryption-zone table is partitioned, the probe's listing and caching are counted on a local hdfs:// filesystem, and the planner cases for the identity gate and the default port. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * feat(iceberg): move the iceberg-rust pin past apache/iceberg-rust#3437 A struct's fields in a data file with neither field ids nor a name mapping now read as NULL, as Spark reads them, whether the struct is read whole or pruned, instead of by position or failing with "cannot reconcile struct fields by id". The test asserts parity with Spark for both reads. What still differs is documented: Spark returns NULL for an optional struct itself where the native scan returns a struct of NULL fields, and an empty list for a list of structs where the native scan keeps the elements with NULL fields. The INT96 migration test asserts those shapes instead of comparing the struct with raw Parquet. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * chore(iceberg): drop the iceberg-rust#3437 note from the pin comment Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(iceberg): create the HDFS encryption zone with the non-deprecated overload The two-argument HdfsAdmin.createEncryptionZone is deprecated in Hadoop 3.3 and 3.4, and the strict-warnings build on the Spark 3.5 profile rejects it. The overload with an empty CreateEncryptionZoneFlag set is what it delegates to. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(iceberg): prove the native HDFS client reads as the Spark user The planner claims an HDFS scan on the premise that iceberg-rust's client connects as the Spark user. MiniDFSCluster enforces permissions by default, but the JVM's user is the superuser of every NameNode it starts, so no test showed it. A cluster started under another user's UGI makes the JVM's user an ordinary one: with the table's data directory at 700 the native scan reads it, and once the directory belongs to someone else the native read is denied with AccessControlException, as Spark's is. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * feat(iceberg): forward the HDFS client settings the session sets to the native reader iceberg-rust's HDFS client reads the executors' $HADOOP_CONF_DIR XML, overridden by the hadoop.* properties of the FileIO, so a client setting the driver holds (spark.hadoop.* or its own XML) never reached it. Of what hdfs-native reads from its configuration, the planner now forwards dfs.client.use.datanode.hostname, dfs.client.socket-timeout and the three dfs.client.failover pacing keys when the session sets them, behind the catalog's own hadoop.* properties. Hadoop's own defaults are not forwarded, since they would override the executors' XML. Not forwarded: the per-nameservice dfs.client.failover.*.<ns> keys, which the client would look up under opendal's own nameservice name, and dfs.data.transfer.protection, on which the client negotiates SASL whenever it is set while the JVM client does so only on a secure cluster, which falls back. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): keep the HDFS encryption-zone listings per parent, with a TTL and a budget The check cached one flag per data directory in a 10000-entry LRU, so a table with more data directories than that listed its parents again on every query, and a cached answer lived for the process: a table purged and re-created inside a new zone at the same path would be claimed and fail on the executors. The listings are now kept per filesystem and parent directory, with every child's flag, so entries scale with parents rather than partitions, and a directory missing from a kept listing, a new partition, lists its parent again. A listing expires with spark.sql.metadataCacheTTLSeconds, Spark's TTL for its own partition file metadata cache (unset by default). A scan that would take more than 1000 listings is declined after that many, which are kept, so a later scan continues from them. The fallback reason now says which of the three happened. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(iceberg): read a merge-on-read table's position deletes over HDFS Delete files are opened through the same FileIO as the data files, and no HDFS test read any; this one writes a position-delete file on HDFS and reads it in one native scan. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): read the executors' HADOOP_USER_NAME from the SparkContext, not the session The HDFS identity gate read spark.executorEnv.HADOOP_USER_NAME from the session's SQLConf first. A session built on an existing context, or a SET, can hold the key there without the executors ever seeing it, and the scan would be claimed while the native client connects as another user, with no fallback. The gate now reads only the SparkContext's configuration, which launched the executors. The planner tests set the live conf instead of the SQLConf, and one pins that the SQLConf value does not stand in. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): keep HDFS encryption-zone listings per table and for an hour by default With spark.sql.metadataCacheTTLSeconds unset, a listing lived for the process, so a table dropped with purge and created again at the same path inside a new encryption zone was still claimed from the old listing and failed on the executors. Listings are now keyed by the table's UUID as well, which the re-created table does not share, so its first scan lists again; and a listing older than an hour is taken again when the TTL is unset, as a bound for anything else. A few minutes would re-list every table of a key-provider cluster constantly and push tables with more than 1000 parents over the listing budget after each expiry. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): give getMetadataLocation its Scaladoc back getTableUuid was inserted between the doc block of getMetadataLocation and its definition, so the paragraph on the metadata-table gate documented the new method and getMetadataLocation had none. Move getTableUuid, with its own doc, above that block. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(iceberg): probe encryption zones in linear time and bound what is kept by entries The probe looked each requested directory up with a scan of the parent's listing, rebuilding Path.toString per comparison: directories times children, billions of comparisons on the driver for a flat table of tens of thousands of partitions, repeated on every cold listing. The lookup is now a map lookup, and children are kept by name rather than full path. Every child of a parent is still kept, so a later scan of other partitions lists nothing, but the whole is now bounded to a million children across tables, least recently used parents evicted first, instead of ten thousand parents of any size. The per-scan budget counts directory entries rather than listStatus calls, a million, which at HDFS's default page of a thousand is the thousand round trips the old budget meant; the listing is paged, so a parent beyond the budget costs the budget and is not kept. The suite probes 50k partitions under one parent: one listing, no per-directory status, under ten seconds, 50k entries kept; the budget declines without keeping, and the kept bound evicts the older table. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * feat(iceberg): add spark.comet.scan.icebergNative.hdfs.enabled The hdfs-native backend is experimental, and the only way back to Spark's Iceberg reader for HDFS tables was to turn every native Iceberg scan off. The new switch, true by default, declines a scan that reads HDFS files with the key in the reason, skips the other HDFS checks and the encryption-zone probe, and leaves scans of other storage alone. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(iceberg): drop the removed Spark 3.4 profile from the HDFS suite's Scaladoc Main no longer has a spark-3.4 profile, so the suite's note on which profiles run it lists 3.5 through 4.2 only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(iceberg): put BouncyCastle on the Spark 4.0 test class path for the HDFS suite On the Spark 4.0 profile the encryption-zone test in CometIcebergHdfsSuite aborted the whole ScalaTest run with NoClassDefFoundError: org/bouncycastle/jce/provider/BouncyCastleProvider Spark 4.0 uses Hadoop 3.4.1, whose KeyProvider constructor links against BouncyCastleProvider, so the class is loaded as soon as a key provider is created. Hadoop 3.4.2 and later load it reflectively and 3.3.4 does not reference it, so the other profiles were unaffected. Add bcprov-jdk18on 1.78.1 (the version Hadoop 3.4.1 builds with) as a test-scoped dependency of the spark-4.0 profile, and note it in the suite's Scaladoc. Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com>

mixermt and others added 2 commits August 29, 2026 18:57

Merge branch 'main' into feat-storage-hdfs-native

4e0f745

blackmwk reviewed Aug 31, 2026

View reviewed changes

Comment thread .github/workflows/ci.yml Resolved

Comment thread crates/iceberg/src/io/storage/config/hdfs.rs Resolved

blackmwk requested changes Aug 31, 2026

View reviewed changes

Comment thread crates/storage/opendal/src/hdfs.rs Resolved

Comment thread crates/storage/opendal/src/lib.rs Resolved

mixermt and others added 4 commits August 31, 2026 12:17

Merge remote-tracking branch 'origin/feat-storage-hdfs-native' into f…

7939b68

…eat-storage-hdfs-native

Merge branch 'main' into feat-storage-hdfs-native

4141f4e

mixermt requested a review from blackmwk September 1, 2026 12:05

Merge branch 'main' into feat-storage-hdfs-native

2578c03

blackmwk reviewed Sep 4, 2026

View reviewed changes

Comment thread crates/iceberg/src/io/storage/config/hdfs.rs Resolved

Comment thread crates/storage/opendal/src/hdfs_native.rs Resolved

Comment thread crates/storage/opendal/src/hdfs.rs Resolved

mixermt and others added 2 commits September 5, 2026 12:11

Merge remote-tracking branch 'apache/main' into feat-storage-hdfs-native

13911c5

# Conflicts: # Cargo.lock

mixermt requested a review from blackmwk September 5, 2026 21:28

mixermt and others added 3 commits September 8, 2026 18:44

Merge branch 'main' into feat-storage-hdfs-native

a9a6786

Merge branch 'main' into feat-storage-hdfs-native

806bd89

ci: retrigger after setup-python manifest timeout on macos runner

9d7d2d8

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

This was referenced Sep 13, 2026

Support Iceberg tables on HDFS in the native scan apache/datafusion-comet#5894

Closed

feat(iceberg): native Iceberg scans of HDFS-backed tables via iceberg-rust's hdfs-native backend apache/datafusion-comet#5898

Merged

comphead reviewed Sep 15, 2026

View reviewed changes

Comment thread dev/docker-compose.yaml Resolved

mixermt and others added 3 commits September 16, 2026 15:58

Merge branch 'main' into feat-storage-hdfs-native

c561cfc

chore(deps): bump rustls to 0.23.45 (RUSTSEC-2026-0285)

b8cc860

Inherited from main (tracked in apache#3222); same bump as apache#3165. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Merge remote-tracking branch 'apache/main' into feat-storage-hdfs-native

587688c

# Conflicts: # Cargo.lock

andygrove mentioned this pull request Sep 21, 2026

deps: bump the iceberg-rust pin to bb1e4a4 and document why it is pinned apache/datafusion-comet#6094

Merged

Merge branch 'main' into feat-storage-hdfs-native

72305b6

Merge branch 'main' into feat-storage-hdfs-native

988be8b

laskoviymishka approved these changes Oct 21, 2026

View reviewed changes

blackmwk reviewed Oct 22, 2026

View reviewed changes

Comment thread crates/storage/opendal/src/hdfs_native.rs Resolved

Comment thread crates/storage/opendal/README.md Resolved

This was referenced Oct 22, 2026

Pass the catalog's Runtime to storage factories #3509

Open

Document how to configure each OpenDAL storage backend #3510

Open

blackmwk approved these changes Oct 22, 2026

View reviewed changes

blackmwk added this pull request to the merge queue Oct 22, 2026

Merged via the queue into apache:main with commit ca21f15 Oct 22, 2026
24 checks passed

comphead mentioned this pull request Oct 22, 2026

feat: read Iceberg tables on HDFS natively through iceberg-rust's hdfs-native backend apache/datafusion-comet#7157

Draft

comphead mentioned this pull request Oct 22, 2026

feat(storage): Add HDFS support via opendal services-hdfs (libhdfs) #3515

Draft

dragonknightk mentioned this pull request Oct 28, 2026

test(storage-opendal): split backend tests into per-backend modules #3569

Open

Conversation

mixermt commented Aug 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

What changes are included in this PR?

Are these changes tested?

AI Disclosure

Uh oh!

Uh oh!

Uh oh!

blackmwk left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

blackmwk left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mixermt commented Sep 5, 2026

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mixermt commented Sep 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mixermt commented Sep 22, 2026

Uh oh!

mixermt commented Oct 10, 2026

Uh oh!

comphead commented Oct 12, 2026

Uh oh!

laskoviymishka left a comment

Choose a reason for hiding this comment

Uh oh!

blackmwk left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

blackmwk left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

comphead commented Oct 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jordepic commented Oct 22, 2026

Uh oh!

laskoviymishka commented Oct 22, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

mixermt commented Aug 30, 2026 •

edited

Loading

mixermt commented Sep 17, 2026 •

edited

Loading

comphead commented Oct 22, 2026 •

edited

Loading