diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml
index 5264a58f676..88e63b9b077 100644
--- a/.github/workflows/pr_build_linux.yml
+++ b/.github/workflows/pr_build_linux.yml
@@ -506,6 +506,7 @@ jobs:
               org.apache.comet.CometVariantProjectionSuite
               org.apache.spark.sql.CometVariantShreddingSuite
               org.apache.comet.CometIcebergNativeSuite
+              org.apache.comet.CometIcebergHdfsSuite
               org.apache.comet.CometBroadcastBuildSideSuite
               org.apache.comet.CometIcebergEncryptionSuite
               org.apache.comet.CometIcebergRewriteActionSuite
diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml
index 55f04de3cd3..4adf8ed22c4 100644
--- a/.github/workflows/pr_build_macos.yml
+++ b/.github/workflows/pr_build_macos.yml
@@ -172,6 +172,7 @@ jobs:
               org.apache.comet.CometVariantProjectionSuite
               org.apache.spark.sql.CometVariantShreddingSuite
               org.apache.comet.CometIcebergNativeSuite
+              org.apache.comet.CometIcebergHdfsSuite
               org.apache.comet.CometBroadcastBuildSideSuite
               org.apache.comet.CometIcebergEncryptionSuite
               org.apache.comet.CometIcebergRewriteActionSuite
diff --git a/docs/source/contributor-guide/roadmap.md b/docs/source/contributor-guide/roadmap.md
index 06bff3435dd..2724910e85f 100644
--- a/docs/source/contributor-guide/roadmap.md
+++ b/docs/source/contributor-guide/roadmap.md
@@ -68,9 +68,8 @@ Comet's native Iceberg scans read V3 tables, including encrypted tables ([#4991]
 ([#5853]). We want to add the remaining V3 features so that these scans don't fall back to Spark: row lineage
 metadata columns, column default values, and the new V3 types (`variant`, `geometry`, `geography`, and `unknown`).
 The work is tracked in [#3376], and upstream `iceberg-rust` support in [iceberg-rust #2411]. Native Iceberg scans
-also don't support HDFS-backed tables today: Comet's native Iceberg storage layer handles only local files, S3 and
-S3-compatible stores, GCS, and OSS, and would need an HDFS `StorageFactory` upstream in `iceberg-storage-opendal`.
-We're scoping what that work would take.
+read HDFS-backed tables through iceberg-rust's experimental `hdfs-native` backend, but fall back to Spark on clusters
+with Kerberos, because that backend cannot authenticate with delegation tokens yet.
 
 [#3376]: https://github.com/apache/datafusion-comet/issues/3376
 [#4991]: https://github.com/apache/datafusion-comet/pull/4991
diff --git a/docs/source/contributor-guide/s3-credential-provider-design.md b/docs/source/contributor-guide/s3-credential-provider-design.md
index a30548566bd..416e2f8e50f 100644
--- a/docs/source/contributor-guide/s3-credential-provider-design.md
+++ b/docs/source/contributor-guide/s3-credential-provider-design.md
@@ -142,7 +142,7 @@ The 5-minute fallback on the Iceberg path bounds how long reqsign reuses a crede
 
 ## Property-bag handling on the Iceberg path
 
-The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The storage-prefix filter (`s3.`/`gcs.`/`adls.`/`client.`/`opendal.`) is applied native-side in `iceberg_common.rs::build_file_io` immediately before `FileIOBuilder.with_prop`. This means the bridge sees `credentials.uri`, OAuth tokens, and any vendor-custom keys with no parallel field on the operator and no driver-side broadcast. Vendors set their own keys on the catalog config and read them back inside `initialize(Map)`.
+The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The storage-prefix filter (`s3.`/`gcs.`/`adls.`/`client.`/`opendal.`/`hdfs.`/`hadoop.`) is applied native-side in `iceberg_common.rs::build_file_io` immediately before `FileIOBuilder.with_prop`. This means the bridge sees `credentials.uri`, OAuth tokens, and any vendor-custom keys with no parallel field on the operator and no driver-side broadcast. Vendors set their own keys on the catalog config and read them back inside `initialize(Map)`.
 
 `IcebergScanExec` derives a redacting `Debug`, and the `FileIO` cache key's `Debug` omits the property bag, so plan dumps and tracing do not leak it.
 
diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md
index b9754029911..862f42c24c3 100644
--- a/docs/source/user-guide/latest/datasources.md
+++ b/docs/source/user-guide/latest/datasources.md
@@ -166,8 +166,8 @@ attaches its own worker threads, so a worker that has read from HDFS can crash t
 the HDFS read itself, typically while an unrelated query is running.
 ```
 
-Native Iceberg scans do not support HDFS-backed tables; those scans fall back to Spark. See the
-[Iceberg](iceberg.md) page.
+Native Iceberg scans read HDFS through a different client, iceberg-rust's `hdfs-native` backend. See
+the [Iceberg](iceberg.md#hdfs) page.
 
 ### Building Comet with HDFS support
 
diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index fd333043013..1dd08092bba 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -138,8 +138,7 @@ safe partial pruning is tracked in [#5883](https://github.com/apache/datafusion-
 - Local filesystem
 - S3-compatible storage (AWS S3, MinIO)
 - Google Cloud Storage (`gs`) and Alibaba Cloud OSS (`oss`)
-
-HDFS-backed tables are not supported by the native Iceberg reader and fall back to Spark.
+- HDFS (`hdfs`), experimental (see [HDFS](#hdfs))
 
 ### REST Catalog
 
@@ -194,6 +193,32 @@ For a custom S3-compatible endpoint, configure the catalog with the endpoint, pa
 
 These `s3.*` storage properties are not specific to the Hive catalog shown here. When `s3.access-key-id` / `s3.secret-access-key` are omitted, credentials come from the standard AWS chain (environment variables, instance profiles, and so on). The region is not auto-detected: when neither the catalog (`client.region` or `s3.region`) nor the executor environment (`AWS_REGION` or `AWS_DEFAULT_REGION`) supplies one, Comet uses `us-east-1`, so set it for AWS buckets in any other region. If your REST catalog vends temporary credentials, the native reader does not consume them automatically, and wiring that requires the credential provider bridge. See Iceberg's [S3 FileIO](https://iceberg.apache.org/docs/latest/aws/#s3-fileio) docs for the full property list, and [S3 Credential Providers](s3-credential-providers.md) for vended or per-request credentials.
 
+### HDFS
+
+The native reader reads HDFS through iceberg-rust's experimental `hdfs-native` backend, a Rust HDFS
+client that uses neither libhdfs nor the JVM. The client reads `core-site.xml` and `hdfs-site.xml`
+from `$HADOOP_CONF_DIR`, or from `$HADOOP_HOME/etc/hadoop`, on each executor, so `spark.hadoop.*`
+settings do not reach it. The one exception is an HA nameservice. A location such as
+`hdfs://ns1/warehouse` has no port, and Comet passes the client the NameNodes that the session's
+Hadoop configuration declares for `ns1` with `dfs.ha.namenodes.ns1` and
+`dfs.namenode.rpc-address.ns1.<id>`. A portless location that the configuration does not declare
+is a single NameNode on Hadoop's default port, 8020, as it is to the JVM client.
+
+A scan of HDFS files falls back to Spark when `hadoop.security.authentication` is not `simple`,
+because the client cannot authenticate with the delegation tokens that Spark executors use on a
+Kerberos cluster. It also falls back when a file location has no port and names a nameservice that
+the Hadoop configuration does not declare, or when it has no authority (`hdfs:///path`), and when a
+data or delete file lies in an HDFS encryption zone: the client is built without KMS support and
+cannot decrypt it. Comet checks that at planning, on clusters whose NameNode advertises a key
+provider, with one directory listing per parent of the data directories, cached for the session.
+
+On a cluster without Kerberos, the client connects as `HADOOP_USER_NAME`, or as the executor's
+operating system user when that variable is unset, rather than as the Spark user. Comet therefore
+claims an HDFS scan only when the executors are known to read as the Spark user: set
+`spark.executorEnv.HADOOP_USER_NAME` to the user that submits the job (in local mode, the JVM's
+`HADOOP_USER_NAME` or operating system user must be that user). Otherwise the scan falls back with
+a reason that names the setting. Writes to HDFS use Iceberg's own writer.
+
 ### Current limitations
 
 The following scenarios will fall back to the JVM Iceberg reader:
@@ -210,6 +235,12 @@ The following scenarios will fall back to the JVM Iceberg reader:
 - Queries that use `input_file_name()`, `input_file_block_start()` or `input_file_block_length()`
 - Scans whose data or delete files span more than one S3 bucket (the native reader uses one
   object-store configuration per scan)
+- HDFS scans on a cluster with Kerberos, or whose executors are not known to read as the Spark
+  user; HDFS file locations that name a nameservice without NameNodes or that have no authority;
+  and HDFS files in an encryption zone (see [HDFS](#hdfs))
+- Scans whose data or delete files are on a different storage backend than the table metadata, for
+  example a local or REST-served metadata location with `write.data.path` on S3 or HDFS. The native
+  reader opens every file of a scan through one storage, chosen from the metadata location
 - Scans where Iceberg reports an output ordering (for example with
   `spark.sql.iceberg.planning.preserve-data-ordering=true`). Spark may then skip the sort above the
   scan, and the native reader does not preserve file order
@@ -800,8 +831,11 @@ Which entries apply depends on which native paths are enabled:
 
 ### Reads that differ from Spark
 
-- **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such
-  a file as NULL, and the native scan returns the values stored in the file
+- **Data files with neither field ids nor a name mapping.** Neither reader can match such a
+  file's nested fields to the table's, so a struct's fields read as NULL in both. The containers
+  differ: Spark's reader 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
+  returns the list with each element's fields NULL
   ([#6794](https://github.com/apache/datafusion-comet/issues/6794)).
 - **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as
   NULL, and two columns that swapped names return each other's values
diff --git a/native/Cargo.lock b/native/Cargo.lock
index 80741b85406..69754efb0cc 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -34,10 +34,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0"
 dependencies = [
  "cfg-if",
- "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.1",
+]
+
 [[package]]
 name = "aes-gcm"
 version = "0.10.3"
@@ -45,9 +56,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",
 ]
@@ -1168,6 +1179,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 = "blocking"
 version = "1.7.0"
@@ -1299,7 +1319,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]]
@@ -1395,7 +1424,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.5",
 ]
 
 [[package]]
@@ -1603,6 +1643,12 @@ dependencies = [
  "cfg-if",
 ]
 
+[[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"
@@ -1621,6 +1667,21 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "crc"
+version = "3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d"
+dependencies = [
+ "crc-catalog",
+]
+
+[[package]]
+name = "crc-catalog"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853"
+
 [[package]]
 name = "crc-fast"
 version = "1.10.0"
@@ -1781,7 +1842,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]]
@@ -2017,7 +2087,7 @@ dependencies = [
  "jni",
  "lazy_static",
  "libc",
- "libloading",
+ "libloading 0.8.9",
  "log",
  "log4rs",
  "mimalloc",
@@ -2876,6 +2946,15 @@ dependencies = [
  "unicode-xid",
 ]
 
+[[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 = "destructure_traitobject"
 version = "0.2.0"
@@ -2926,6 +3005,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",
+ "libc",
+ "socket2",
+ "windows-sys 0.60.2",
+]
+
 [[package]]
 name = "dunce"
 version = "1.0.5"
@@ -3214,6 +3305,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 2.0.119",
+]
+
+[[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 = "generic-array"
 version = "0.14.7"
@@ -3233,7 +3352,7 @@ dependencies = [
  "cfg-if",
  "js-sys",
  "libc",
- "wasi",
+ "wasi 0.11.1+wasi-snapshot-preview1",
  "wasm-bindgen",
 ]
 
@@ -3360,6 +3479,48 @@ dependencies = [
  "foldhash 0.2.0",
 ]
 
+[[package]]
+name = "hdfs-native"
+version = "0.14.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5141c4b745cab57a0018daa9c2759c92308bcf4e41e0c617d05fbf9639c16754"
+dependencies = [
+ "aes 0.9.3",
+ "base64 0.22.1",
+ "bitflags 2.13.2",
+ "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 0.9.0",
+ "log",
+ "md-5 0.11.0",
+ "num-traits",
+ "once_cell",
+ "prost",
+ "prost-types",
+ "rand 0.10.3",
+ "regex",
+ "roxmltree",
+ "socket2",
+ "thiserror",
+ "tokio",
+ "tokio-util",
+ "url",
+ "uuid",
+ "whoami",
+]
+
 [[package]]
 name = "hdfs-sys"
 version = "0.3.1"
@@ -3568,7 +3729,7 @@ dependencies = [
 [[package]]
 name = "iceberg"
 version = "0.10.1"
-source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b"
+source = "git+https://github.com/apache/iceberg-rust?rev=de0a6ac8471da43f63382d66fac27d7e56fd273e#de0a6ac8471da43f63382d66fac27d7e56fd273e"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -3626,7 +3787,7 @@ dependencies = [
 [[package]]
 name = "iceberg-property-macro"
 version = "0.10.1"
-source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b"
+source = "git+https://github.com/apache/iceberg-rust?rev=de0a6ac8471da43f63382d66fac27d7e56fd273e#de0a6ac8471da43f63382d66fac27d7e56fd273e"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3636,7 +3797,7 @@ dependencies = [
 [[package]]
 name = "iceberg-storage-opendal"
 version = "0.10.1"
-source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b"
+source = "git+https://github.com/apache/iceberg-rust?rev=de0a6ac8471da43f63382d66fac27d7e56fd273e#de0a6ac8471da43f63382d66fac27d7e56fd273e"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3649,6 +3810,8 @@ dependencies = [
  "reqsign-aws-v4",
  "reqsign-core",
  "serde",
+ "tokio",
+ "tracing",
  "typetag",
  "url",
 ]
@@ -3810,10 +3973,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.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b5cf4372d022092674a83b8febf5865f7a573cc0c5baec5e49c879b6a20978f"
+dependencies = [
+ "block-padding 0.4.2",
+ "hybrid-array",
+]
+
 [[package]]
 name = "inventory"
 version = "0.3.25"
@@ -3955,7 +4128,7 @@ dependencies = [
  "java-locator",
  "jni-macros",
  "jni-sys",
- "libloading",
+ "libloading 0.8.9",
  "log",
  "simd_cesu8",
  "thiserror",
@@ -4113,6 +4286,16 @@ dependencies = [
  "windows-link",
 ]
 
+[[package]]
+name = "libloading"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "754ca22de805bb5744484a5b151a9e1a8e837d5dc232c2d7d8c2e3492edc8b60"
+dependencies = [
+ "cfg-if",
+ "windows-link",
+]
+
 [[package]]
 name = "liblzma"
 version = "0.4.9"
@@ -4148,6 +4331,15 @@ dependencies = [
  "cc",
 ]
 
+[[package]]
+name = "libredox"
+version = "0.1.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "61ff90caf6077a803a240f62fdbe88645a890bbca49ef8174c3cb0404362171d"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "link-section"
 version = "0.19.3"
@@ -4321,7 +4513,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee"
 dependencies = [
  "libc",
- "wasi",
+ "wasi 0.11.1+wasi-snapshot-preview1",
  "windows-sys 0.61.2",
 ]
 
@@ -4490,6 +4682,24 @@ dependencies = [
  "libm",
 ]
 
+[[package]]
+name = "objc2-core-foundation"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536"
+dependencies = [
+ "bitflags 2.13.2",
+]
+
+[[package]]
+name = "objc2-system-configuration"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396"
+dependencies = [
+ "objc2-core-foundation",
+]
+
 [[package]]
 name = "object"
 version = "0.37.3"
@@ -4597,6 +4807,7 @@ dependencies = [
  "opendal-service-fs",
  "opendal-service-gcs",
  "opendal-service-hdfs",
+ "opendal-service-hdfs-native",
  "opendal-service-oss",
  "opendal-service-s3",
 ]
@@ -4765,6 +4976,20 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "opendal-service-hdfs-native"
+version = "0.58.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8aebbf956ea9e64d70fc8d1f25381604c02528805c23bd55443b8e924b438222"
+dependencies = [
+ "bytes",
+ "futures",
+ "hdfs-native",
+ "log",
+ "opendal-core",
+ "serde",
+]
+
 [[package]]
 name = "opendal-service-oss"
 version = "0.58.2"
@@ -5112,8 +5337,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",
@@ -5962,6 +6187,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"
@@ -6121,7 +6355,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213"
 dependencies = [
- "cipher",
+ "cipher 0.4.4",
 ]
 
 [[package]]
@@ -7180,6 +7414,15 @@ version = "0.11.1+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
+[[package]]
+name = "wasi"
+version = "0.14.7+wasi-0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c"
+dependencies = [
+ "wasip2",
+]
+
 [[package]]
 name = "wasip2"
 version = "1.0.4+wasi-0.2.12"
@@ -7189,6 +7432,15 @@ dependencies = [
  "wit-bindgen",
 ]
 
+[[package]]
+name = "wasite"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "66fe902b4a6b8028a753d5424909b764ccf79b7a209eac9bf97e59cda9f71a42"
+dependencies = [
+ "wasi 0.14.7+wasi-0.2.4",
+]
+
 [[package]]
 name = "wasm-bindgen"
 version = "0.2.129"
@@ -7300,6 +7552,19 @@ dependencies = [
  "rustls-pki-types",
 ]
 
+[[package]]
+name = "whoami"
+version = "2.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "626c4bac6755d76ffc12cb01b2eac751db1996b9e0041de9aa02c8c211ddc82c"
+dependencies = [
+ "libc",
+ "libredox",
+ "objc2-system-configuration",
+ "wasite",
+ "web-sys",
+]
+
 [[package]]
 name = "winapi"
 version = "0.3.9"
@@ -7396,7 +7661,16 @@ version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
 dependencies = [
- "windows-targets",
+ "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]]
@@ -7414,14 +7688,31 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
 dependencies = [
- "windows_aarch64_gnullvm",
- "windows_aarch64_msvc",
- "windows_i686_gnu",
- "windows_i686_gnullvm",
- "windows_i686_msvc",
- "windows_x86_64_gnu",
- "windows_x86_64_gnullvm",
- "windows_x86_64_msvc",
+ "windows_aarch64_gnullvm 0.52.6",
+ "windows_aarch64_msvc 0.52.6",
+ "windows_i686_gnu 0.52.6",
+ "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]]
@@ -7430,48 +7721,96 @@ 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.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.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.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.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.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.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 = "wit-bindgen"
 version = "0.57.1"
diff --git a/native/Cargo.toml b/native/Cargo.toml
index c858f9e715b..24cd6e3fb69 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -85,8 +85,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false }
 # Pinned to a revision rather than a release because no published iceberg-rust
 # version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain
 # version requirement once a release ships on arrow 59. Bump policy: #5645.
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "de0a6ac8471da43f63382d66fac27d7e56fd273e" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "de0a6ac8471da43f63382d66fac27d7e56fd273e", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls", "opendal-hdfs-native"] }
 reqsign-core = "3"
 
 [profile.release]
diff --git a/native/core/src/execution/operators/iceberg_common.rs b/native/core/src/execution/operators/iceberg_common.rs
index a17eb076e0d..f3f2286bb8a 100644
--- a/native/core/src/execution/operators/iceberg_common.rs
+++ b/native/core/src/execution/operators/iceberg_common.rs
@@ -45,8 +45,11 @@ const ICEBERG_PROVIDER_CLASS_PROPERTY: &str = "s3.comet.credential.provider.clas
 /// Key prefixes forwarded to iceberg-rust's `FileIO`. The full unfiltered catalog bag (catalog
 /// URI, OAuth tokens, credentials.uri, tenant-id, etc.) is kept upstream so
 /// `CometS3CredentialBridge` can read whatever the vendor needs. `opendal.` carries
-/// iceberg-storage-opendal's own settings, such as `opendal.io-timeout-ms`.
-const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client.", "opendal."];
+/// iceberg-storage-opendal's own settings, such as `opendal.io-timeout-ms`. `hdfs.` and `hadoop.`
+/// carry the HDFS backend's NameNode declarations and Hadoop client settings.
+const STORAGE_PROPERTY_PREFIXES: &[&str] = &[
+    "s3.", "gcs.", "adls.", "client.", "opendal.", "hdfs.", "hadoop.",
+];
 
 /// Pick an OpenDAL storage backend for a URI whose scheme `builtin_storage_schemes` lists for
 /// `access_mode`, or that is an opted-in S3-compliant alias written in lowercase; anything else is
@@ -79,6 +82,7 @@ pub(crate) fn storage_factory_for(
         "memory" => Ok((Arc::new(OpenDalStorageFactory::Memory), true)),
         "gs" => Ok((Arc::new(OpenDalStorageFactory::Gcs), true)),
         "oss" => Ok((Arc::new(OpenDalStorageFactory::Oss), true)),
+        "hdfs" => Ok((Arc::new(OpenDalStorageFactory::HdfsNative), true)),
         // s3, s3a, and any opted-in s3-compliant alias (e.g. blob) route to the S3 backend. Listed
         // last so the built-in backends above stay authoritative even if one of their schemes is
         // also named in `fs.comet.s3Compliant.schemes`. An alias additionally gets a wrapper that
@@ -302,9 +306,10 @@ fn cached_file_io(
 /// mode; the JVM read and write gates load these lists over JNI. `memory` is write-only: an OpenDAL
 /// memory backend is a fresh empty in-process store the write path assembles manifests in, so a
 /// read finds nothing. `oss` is read-only: no `oss.*` property is forwarded and no test covers it.
+/// `hdfs` is read-only: the write path has not been tested against it.
 pub(crate) fn builtin_storage_schemes(access_mode: AccessMode) -> &'static [&'static str] {
     match access_mode {
-        AccessMode::Read => &["file", "gs", "oss", "s3", "s3a"],
+        AccessMode::Read => &["file", "gs", "hdfs", "oss", "s3", "s3a"],
         AccessMode::Write => &["file", "memory", "gs", "s3", "s3a"],
     }
 }
@@ -847,6 +852,158 @@ mod tests {
         );
     }
 
+    #[test]
+    fn hdfs_scheme_is_readable_but_not_writable() {
+        assert!(factory_result("hdfs://nn:8020/db/table", AccessMode::Read).is_ok());
+        let err = factory_result("hdfs://nn:8020/db/table", AccessMode::Write).unwrap_err();
+        assert!(
+            err.contains("Unsupported storage scheme: hdfs"),
+            "unexpected error: {err}"
+        );
+    }
+
+    /// The JVM declares a nameservice as `hdfs.name-node.<ns>`, and a catalog can declare one with
+    /// Hadoop's own keys under `hadoop.`.
+    #[test]
+    fn hdfs_properties_reach_the_file_io() {
+        let props = HashMap::from([
+            ("hdfs.name-node.ns1".to_string(), "nn1:8020".to_string()),
+            ("hadoop.dfs.ha.namenodes.ns2".to_string(), "a".to_string()),
+            (
+                "hadoop.dfs.namenode.rpc-address.ns2.a".to_string(),
+                "nn2:8020".to_string(),
+            ),
+        ]);
+        let (file_io, _) =
+            build_file_io(&props, "hdfs://ns2/warehouse", "", AccessMode::Read).unwrap();
+        for (key, value) in &props {
+            assert_eq!(file_io.config().get(key), Some(value), "{key}");
+        }
+    }
+
+    /// The cache is shared per executor, and an HDFS `FileIO` keeps its NameNode clients for life.
+    /// Two catalogs that differ only in how they reach HDFS must therefore never share a key, or
+    /// the second would be served by the first one's clients. A regression pin on the key, which
+    /// hashes every property.
+    #[test]
+    fn cache_key_separates_hdfs_catalogs() {
+        let table = "hdfs://nameservice1/warehouse/db/t";
+        let props = |name_node: &str, random_order: &str| {
+            HashMap::from([
+                // What the JVM declares for an `hdfs://nameservice1/...` table.
+                (
+                    "hdfs.name-node.nameservice1".to_string(),
+                    name_node.to_string(),
+                ),
+                (
+                    "hadoop.dfs.client.failover.random.order".to_string(),
+                    random_order.to_string(),
+                ),
+            ])
+        };
+        let key = |props: &HashMap<String, String>| {
+            FileIoCacheKey::new(props, table, "cat", AccessMode::Read)
+        };
+        let base = props("nn1:8020,nn2:8020", "true");
+
+        // Control: the same properties always give the same key, so the inequalities below are
+        // down to the one property that changed.
+        assert!(key(&base).is_some());
+        assert_eq!(key(&base), key(&base.clone()));
+
+        // Only the declaration differs: another NameNode, another list, the same list reordered
+        // (which changes the failover order), and no list at all.
+        for other in [
+            props("nn1:8020,nn3:8020", "true"),
+            props("nn1:8020", "true"),
+            props("nn2:8020,nn1:8020", "true"),
+        ] {
+            assert_ne!(key(&base), key(&other), "{other:?}");
+        }
+        let mut without_name_node = base.clone();
+        without_name_node.remove("hdfs.name-node.nameservice1");
+        assert_ne!(key(&base), key(&without_name_node));
+
+        // Only the `hadoop.dfs.client.failover.*` overrides differ: a changed value, an added
+        // override and a removed one.
+        let mut other_failover = base.clone();
+        other_failover.insert(
+            "hadoop.dfs.client.failover.proxy.provider.nameservice1".to_string(),
+            "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider".to_string(),
+        );
+        let mut without_failover = base.clone();
+        without_failover.remove("hadoop.dfs.client.failover.random.order");
+        for other in [
+            props("nn1:8020,nn2:8020", "false"),
+            other_failover,
+            without_failover,
+        ] {
+            assert_ne!(key(&base), key(&other), "{other:?}");
+        }
+    }
+
+    /// Builds the real HDFS storage from one property the way a `FileIO` does on first use.
+    /// Parsing is all it does: an operator, and with it a NameNode connection, is only made by
+    /// the first storage call, so this needs neither a NameNode nor a runtime. The storage is
+    /// inspected through its serde form, so this is a contract test on the pinned iceberg-rust
+    /// revision and its config field names.
+    fn build_hdfs_storage(key: &str, value: &str) -> Result<serde_json::Value, String> {
+        let config =
+            StorageConfig::from_props(HashMap::from([(key.to_string(), value.to_string())]));
+        let storage = OpenDalStorageFactory::HdfsNative
+            .build(&config)
+            .map_err(|e| e.to_string())?;
+        Ok(serde_json::to_value(&*storage).expect("the storage serializes"))
+    }
+
+    /// One Hadoop option of the parsed storage config.
+    fn parsed_option<'a>(storage: &'a serde_json::Value, key: &str) -> Option<&'a str> {
+        storage
+            .pointer("/HdfsNative/config/options")
+            .and_then(|options| options.get(key))
+            .and_then(serde_json::Value::as_str)
+    }
+
+    /// The contract `CometIcebergNativeScan.hadoopToIcebergHdfsProperties` relies on: it declares
+    /// an HA nameservice as `hdfs.name-node.<nameservice>`, and iceberg-rust expands that into
+    /// Hadoop's own `dfs.ha.namenodes.<ns>` / `dfs.namenode.rpc-address.<ns>.<nn>`, the keys its
+    /// resolver reads for a portless path authority. A portless entry is refused when the
+    /// storage is built, naming the key and the entry.
+    #[test]
+    fn hdfs_nameservice_declaration_is_parsed_by_the_real_storage() {
+        for input in [
+            "nn1:8020,nn2:8020",
+            // Whitespace around entries, a trailing slash and the `hdfs://` prefix are tolerated.
+            " hdfs://nn1:8020/ , nn2:8020 ",
+        ] {
+            let storage = build_hdfs_storage("hdfs.name-node.ns1", input)
+                .unwrap_or_else(|e| panic!("`{input}` should be accepted: {e}"));
+            assert_eq!(
+                parsed_option(&storage, "dfs.ha.namenodes.ns1"),
+                Some("nn0,nn1"),
+                "input `{input}`: {storage}"
+            );
+            assert_eq!(
+                parsed_option(&storage, "dfs.namenode.rpc-address.ns1.nn0"),
+                Some("nn1:8020"),
+                "input `{input}`: {storage}"
+            );
+            assert_eq!(
+                parsed_option(&storage, "dfs.namenode.rpc-address.ns1.nn1"),
+                Some("nn2:8020"),
+                "input `{input}`: {storage}"
+            );
+        }
+
+        // A portless entry would only fail at the first I/O, so it is refused here, and the
+        // error names the offending key and entry rather than the whole list.
+        let err = build_hdfs_storage("hdfs.name-node.ns1", "nn1:8020,nn2").unwrap_err();
+        assert!(
+            err.contains("hdfs.name-node.ns1") && err.contains("nn2") && !err.contains("nn1:8020"),
+            "unexpected error: {err}"
+        );
+    }
+
     #[test]
     fn memory_scheme_is_writable_but_not_readable() {
         // The write path assembles manifests in a fresh in-process memory store. A read against
@@ -948,7 +1105,7 @@ mod tests {
 
     #[test]
     fn unknown_scheme_is_rejected() {
-        let err = factory_result("hdfs://nn/db/table", AccessMode::Read).unwrap_err();
+        let err = factory_result("viewfs://cluster/db/table", AccessMode::Read).unwrap_err();
         assert!(
             err.contains("Unsupported storage scheme"),
             "unexpected error: {err}"
@@ -997,7 +1154,7 @@ mod tests {
     #[test]
     fn unlisted_schemes_are_rejected_for_both_modes() {
         let unlisted = [
-            "hdfs", "abfs", "abfss", "wasb", "wasbs", "gcs", "http", "https", "azure",
+            "viewfs", "abfs", "abfss", "wasb", "wasbs", "gcs", "http", "https", "azure",
         ];
         for mode in [AccessMode::Read, AccessMode::Write] {
             for scheme in unlisted {
diff --git a/native/core/src/execution/operators/iceberg_partition_path.rs b/native/core/src/execution/operators/iceberg_partition_path.rs
index c9ead454b0f..dba99bc2b6f 100644
--- a/native/core/src/execution/operators/iceberg_partition_path.rs
+++ b/native/core/src/execution/operators/iceberg_partition_path.rs
@@ -152,7 +152,7 @@ const NULL: &str = "null";
 /// | `timestamp`        | `1969-12-31T23:59:58.5`       | `1969-12-31 23:59:58.500`        |
 /// | `timestamptz`      | `1969-12-31T23:59:58.5+00:00` | panics for a negative value with a sub-second part; otherwise `1969-12-31 23:59:58.500 UTC` |
 /// | `binary` / `fixed` | base64                        | uppercase hex                    |
-/// | `float` / `double` | `1.0`, `1.0E20`               | `1`, `100000000000000000000`     |
+/// | `float` / `double` | `1.0`, `1.0E20`               | `1.0`, `1.0E20` since #3358; `Double::MAX` still renders as 309 digits (see below) |
 ///
 /// The nanosecond timestamp types get the same treatment for the same reason. Spark cannot plan a
 /// write to them today, and `CometIcebergNativeWrite`'s column-type rule declines them as well; the
diff --git a/pom.xml b/pom.xml
index 292fc21d702..b8c777e7a57 100644
--- a/pom.xml
+++ b/pom.xml
@@ -87,7 +87,10 @@ under the License.
     <protobuf.version>3.25.5</protobuf.version>
     <parquet.version>1.16.0</parquet.version>
     <parquet.maven.scope>provided</parquet.maven.scope>
-    <hadoop.version>3.3.4</hadoop.version>
+    <!-- hadoop-client-minicluster, test scope. Follows each Spark profile's Hadoop line below, as
+         hadoop-azure.version does: MiniDFSCluster cannot start when the minicluster is older than
+         the hadoop-client-api/runtime Spark supplies. The default matches spark-4.1. -->
+    <hadoop.version>3.4.2</hadoop.version>
     <arrow.version>18.3.0</arrow.version>
     <codehaus.jackson.version>1.9.13</codehaus.jackson.version>
     <spotless.version>2.43.0</spotless.version>
@@ -106,7 +109,7 @@ under the License.
     <!-- hadoop-aws, provided scope, to compile the built-in S3 credential adapters against the S3A
          provider factory. Default tracks the default shim source set (spark-4.x -> SDK v2, needs
          Hadoop 3.4.x's CredentialProviderListFactory); the spark-3.4 / spark-3.5 profiles override
-         it down to 3.3.x (SDK v1). hadoop.version itself (hadoop-client) is not changed. -->
+         it down to 3.3.x (SDK v1). -->
     <hadoop-aws.version>3.4.1</hadoop-aws.version>
     <!-- Delta Lake pairing for the contrib/delta-spark module (-Pdelta). Each Spark profile
          in this pom sets delta.spark.version to its matching Delta release. Only spark-3.4
@@ -731,6 +734,7 @@ under the License.
         <spark.version>3.4.3</spark.version>
         <spark.version.short>3.4</spark.version.short>
         <hadoop-aws.version>3.3.4</hadoop-aws.version>
+        <hadoop.version>3.3.4</hadoop.version>
         <hadoop-azure.version>3.3.4</hadoop-azure.version>
         <parquet.version>1.13.1</parquet.version>
         <semanticdb.version>4.8.8</semanticdb.version>
@@ -757,6 +761,7 @@ under the License.
         <spark.version>3.5.9</spark.version>
         <spark.version.short>3.5</spark.version.short>
         <hadoop-aws.version>3.3.4</hadoop-aws.version>
+        <hadoop.version>3.3.4</hadoop.version>
         <hadoop-azure.version>3.3.4</hadoop-azure.version>
         <parquet.version>1.13.1</parquet.version>
         <semanticdb.version>4.8.8</semanticdb.version>
@@ -778,6 +783,7 @@ under the License.
         <scala.binary.version>2.13</scala.binary.version>
         <spark.version>4.0.4</spark.version>
         <spark.version.short>4.0</spark.version.short>
+        <hadoop.version>3.4.1</hadoop.version>
         <hadoop-azure.version>3.4.1</hadoop-azure.version>
         <parquet.version>1.15.2</parquet.version>
         <semanticdb.version>4.13.10</semanticdb.version>
@@ -802,6 +808,7 @@ under the License.
         <scala.binary.version>2.13</scala.binary.version>
         <spark.version>4.1.3</spark.version>
         <spark.version.short>4.1</spark.version.short>
+        <hadoop.version>3.4.2</hadoop.version>
         <hadoop-azure.version>3.4.2</hadoop-azure.version>
         <parquet.version>1.16.0</parquet.version>
         <semanticdb.version>4.13.10</semanticdb.version>
@@ -826,6 +833,7 @@ under the License.
         <scala.binary.version>2.13</scala.binary.version>
         <spark.version>4.2.0</spark.version>
         <spark.version.short>4.2</spark.version.short>
+        <hadoop.version>3.5.0</hadoop.version>
         <hadoop-azure.version>3.5.0</hadoop-azure.version>
         <parquet.version>1.17.0</parquet.version>
         <semanticdb.version>4.13.10</semanticdb.version>
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index 349a9dae380..7db6f15e19f 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -640,6 +640,12 @@ case class CometScanRule(session: SparkSession)
                 hadoopS3Options,
                 icebergDataBucket)
 
+            // Declares the HDFS nameservices the data/delete files name. Merged before the
+            // catalog's properties, so a catalog's own declaration wins.
+            val hdfsProperties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties(
+              hadoopConf,
+              taskValidation.hdfsAuthorities)
+
             // Forward the full FileIO property bag (including credentials.uri, OAuth tokens,
             // tenant-id, etc.) so a CometS3CredentialProvider can see everything LoadTableResponse
             // returned. The storage-prefix narrowing happens native-side just before
@@ -655,10 +661,17 @@ case class CometScanRule(session: SparkSession)
             // iceberg-rust's FileIO, so the alias key never reaches FileIO -- but it is still
             // handed, unfiltered, to CometS3CredentialBridge, so a custom credential provider sees
             // it. That is intended: the provider gets the full property bag.
-            val catalogProperties = hadoopDerivedProperties ++ fileIOProperties ++
-              hadoopS3Options
-                .get(COMET_S3_COMPLIANT_SCHEMES_KEY)
-                .map(COMET_S3_COMPLIANT_SCHEMES_KEY -> _)
+            // A nameservice the catalog declares with Hadoop's own keys is not also given a
+            // default-port host declaration.
+            val derivedHdfsProperties = hdfsProperties.filterNot { case (key, _) =>
+              val ns = key.stripPrefix("hdfs.name-node.")
+              fileIOProperties.contains(s"hadoop.dfs.ha.namenodes.$ns")
+            }
+            val catalogProperties =
+              hadoopDerivedProperties ++ derivedHdfsProperties ++ fileIOProperties ++
+                hadoopS3Options
+                  .get(COMET_S3_COMPLIANT_SCHEMES_KEY)
+                  .map(COMET_S3_COMPLIANT_SCHEMES_KEY -> _)
 
             val result = CometIcebergNativeScanMetadata.extract(
               scanExec.scan,
@@ -703,6 +716,22 @@ case class CometScanRule(session: SparkSession)
             false
           }
 
+        // The native scan opens every data and delete file through one storage, built from the
+        // metadata location's scheme (`load_file_io` in iceberg_scan.rs), so a file on another
+        // backend fails at open time: an `hdfs://` file handed to the local storage, a local path
+        // to the S3 one. Declined here instead; the multi-bucket check above is the same idea one
+        // level down.
+        val metadataFamily = CometScanRule.storageFamily(metadataUri.getScheme)
+        val foreignFamilies =
+          taskValidation.fileSchemes.map(CometScanRule.storageFamily) - metadataFamily
+        val storageFamiliesConsistent = foreignFamilies.isEmpty || {
+          fallbackReasons += "Iceberg scan reads data/delete files on storage " +
+            s"${foreignFamilies.toSeq.sorted.map(f => s"'$f'").mkString(", ")} while the table " +
+            s"metadata is on '$metadataFamily'; Comet's native Iceberg reader opens every file " +
+            "of a scan through one storage backend, chosen from the metadata location"
+          false
+        }
+
         // Now perform all validation using the pre-extracted metadata
         // Check if table uses a FileIO implementation compatible with iceberg-rust.
         // Comet's native reader bypasses Iceberg Java's FileIO entirely and reads through
@@ -912,6 +941,69 @@ case class CometScanRule(session: SparkSession)
           case None => true
         }
 
+        // iceberg-rust's HDFS client cannot use an executor's delegation tokens: opendal builds
+        // every client for a synthetic nameservice, and the client looks tokens up by that name.
+        // A portless authority must be a declared nameservice (see hadoopToIcebergHdfsProperties).
+        val hdfsSupported = taskValidation.hdfsAuthorities.isEmpty || {
+          val authentication = hadoopConf.getTrimmed("hadoop.security.authentication", "simple")
+          val secure = !authentication.equalsIgnoreCase("simple")
+          if (secure) {
+            fallbackReasons += "Iceberg scan reads HDFS with " +
+              s"hadoop.security.authentication=$authentication, but Comet's native Iceberg " +
+              "reader cannot authenticate to HDFS with delegation tokens"
+          }
+          val undeclared = CometIcebergNativeScan.undeclaredHdfsNameservice(
+            taskValidation.hdfsAuthorities,
+            metadata.catalogProperties)
+          undeclared.foreach { ns =>
+            fallbackReasons += s"Iceberg scan reads HDFS authority '$ns', which has no port and " +
+              "which the Hadoop configuration does not declare as a nameservice " +
+              s"(dfs.ha.namenodes.$ns, and a host:port dfs.namenode.rpc-address.$ns.<id> for " +
+              "each NameNode)"
+          }
+          // The client connects as HADOOP_USER_NAME, or as the executor's OS user, never as the
+          // Spark user the JVM reader runs as (opendal exposes no user setting). Claimed only
+          // when the executors are known to read as the Spark user: HADOOP_USER_NAME set to it
+          // in their environment, or, in local mode, the JVM's own user.
+          val sparkUser = session.sparkContext.sparkUser
+          val executorEnvKey = "spark.executorEnv.HADOOP_USER_NAME"
+          val executorUser = Option(
+            session.sessionState.conf
+              .getConfString(
+                executorEnvKey,
+                session.sparkContext.getConf.get(executorEnvKey, null)))
+            .orElse {
+              if (session.sparkContext.isLocal) {
+                Some(sys.env.getOrElse("HADOOP_USER_NAME", System.getProperty("user.name")))
+              } else {
+                None
+              }
+            }
+          val identityKnown = executorUser.contains(sparkUser)
+          if (!identityKnown) {
+            fallbackReasons += "Iceberg scan reads HDFS as the executors' user " +
+              s"${executorUser.map(u => s"'$u'").getOrElse("(unknown)")} rather than the Spark " +
+              s"user '$sparkUser'; set $executorEnvKey=$sparkUser so Comet's native Iceberg " +
+              "reader connects as the Spark user"
+          }
+          // The client is built without KMS support and rejects every file of an encryption
+          // zone. Probed only once the scan is otherwise eligible, so an undeclared nameservice
+          // is not also dialed.
+          val encryptedDirectory =
+            if (secure || undeclared.nonEmpty || !identityKnown) None
+            else {
+              CometIcebergNativeScan.hdfsEncryptedDirectory(
+                hadoopConf,
+                taskValidation.hdfsDirectories)
+            }
+          encryptedDirectory.foreach { dir =>
+            fallbackReasons += s"Iceberg scan reads HDFS directory '$dir', which is in an " +
+              "encryption zone or whose status could not be read; Comet's native Iceberg reader " +
+              "cannot decrypt HDFS encryption zones"
+          }
+          !secure && undeclared.isEmpty && identityKnown && encryptedDirectory.isEmpty
+        }
+
         if (!taskValidation.allParquet) {
           fallbackReasons += "Iceberg scan contains non-Parquet files (ORC or Avro). " +
             "Comet only supports Parquet files in Iceberg tables"
@@ -1123,9 +1215,11 @@ case class CometScanRule(session: SparkSession)
         if (schemaSupported && fileIOCompatible && formatVersionSupported &&
           defaultValuesSupported && schemaTypesSupported && encryptionKeyLengthSupported &&
           taskValidation.allParquet && allSupportedFilesystems && allLocationsOpenable &&
-          metadataSchemeSupported && partitionTypesSupported && unifiedPartitionTypeSupported &&
+          metadataSchemeSupported && storageFamiliesConsistent && partitionTypesSupported &&
+          unifiedPartitionTypeSupported &&
           transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported &&
-          nestedFieldsSupported && taskSchemaNamesSupported && nestedNameMappingSupported) {
+          nestedFieldsSupported && taskSchemaNamesSupported && hdfsSupported &&
+          nestedNameMappingSupported) {
           CometBatchScanExec(
             scanExec.clone().asInstanceOf[BatchScanExec],
             runtimeFilters = scanExec.runtimeFilters,
@@ -1411,6 +1505,7 @@ object CometScanRule extends Logging {
    * the bucket from the first path segment (`s3_blob_fs_support.rs`), so a hostless
    * `blob:///bucket/key.parquet` IS openable when it carries a promotable bucket segment.
    */
+  /** Whether `dataType` is, or anywhere contains, an array or a map. */
   private[rules] def hasOpenableAuthority(uri: URI, s3CompliantSchemes: Set[String]): Boolean = {
     val scheme = NativeConfig.lowerScheme(uri)
     scheme.isEmpty || scheme.contains("file") || uri.getRawAuthority != null ||
@@ -1463,6 +1558,13 @@ object CometScanRule extends Logging {
     // Buckets the native FileIO must read (data + delete files), for the single-config check in
     // CometScanRule; only S3-family locations contribute (see NativeConfig.bucketForUri).
     val dataFileBuckets = mutable.Set[String]()
+    // Schemes of the data/delete files the native FileIO opens, for the storage-family check in
+    // CometScanRule; a schemeless path is local, as iceberg-rust's `scheme_of` reads it.
+    val fileSchemes = mutable.Set[String]()
+    // HDFS authorities and directories of the data/delete files, for the HDFS checks in
+    // CometScanRule.
+    val hdfsAuthorities = mutable.Set[String]()
+    val hdfsDirectories = mutable.Set[String]()
     // First data/delete location with a readable scheme but no URL host (see
     // hasOpenableAuthority); non-empty => decline. One example suffices for the message.
     var hostlessLocation: Option[String] = None
@@ -1476,14 +1578,24 @@ object CometScanRule extends Logging {
         catch { case _: java.net.URISyntaxException => return }
       // A schemeless local path routes to iceberg-rust's LocalFs and needs no host.
       val scheme = uri.getScheme
-      if (scheme == null) return
+      if (scheme == null) {
+        fileSchemes += "file"
+        return
+      }
       if (!isAdmittedIcebergScheme(scheme, s3CompliantSchemes)) {
         unsupportedSchemes += scheme
       } else if (!hasOpenableAuthority(uri, s3CompliantSchemes)) {
         if (hostlessLocation.isEmpty) hostlessLocation = Some(rawPath)
       } else {
-        // bucketForUri yields None for non-S3-family URIs, so no scheme re-check is needed here.
-        NativeConfig.bucketForUri(uri, s3CompliantSchemes).foreach(dataFileBuckets += _)
+        fileSchemes += scheme
+        if (scheme == "hdfs") {
+          hdfsAuthorities += uri.getRawAuthority
+          val slash = rawPath.lastIndexOf('/')
+          if (slash > 0) hdfsDirectories += rawPath.substring(0, slash)
+        } else {
+          // bucketForUri yields None for non-S3-family URIs, so no scheme re-check is needed.
+          NativeConfig.bucketForUri(uri, s3CompliantSchemes).foreach(dataFileBuckets += _)
+        }
       }
     }
 
@@ -1555,9 +1667,23 @@ object CometScanRule extends Logging {
       nonIdentityTransform,
       deleteFiles,
       dataFileBuckets.toSet,
+      fileSchemes.toSet,
+      hdfsAuthorities.toSet,
+      hdfsDirectories.toSet,
       hostlessLocation,
       requiredFieldIds)
   }
+
+  /**
+   * The native storage backend a data/delete file scheme routes to (`storage_factory_for` in
+   * iceberg_common.rs): `s3` and `s3a` share one, a schemeless path is local, and every other
+   * scheme, an S3-compliant alias included, is its own.
+   */
+  private[rules] def storageFamily(scheme: String): String = scheme match {
+    case null | "file" => "file"
+    case "s3" | "s3a" => "s3"
+    case other => other
+  }
 }
 
 /**
@@ -1569,5 +1695,8 @@ case class IcebergTaskValidationResult(
     nonIdentityTransform: Option[String],
     deleteFiles: java.util.List[_],
     dataFileBuckets: Set[String],
+    fileSchemes: Set[String],
+    hdfsAuthorities: Set[String],
+    hdfsDirectories: Set[String],
     hostlessLocation: Option[String],
     requiredFieldIds: Set[Int])
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index 4070d785367..81df1db2807 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -26,10 +26,14 @@ import java.util.{Base64, UUID}
 
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
 
 import org.json4s.JsonDSL._
 import org.json4s.jackson.JsonMethods._
 
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.comet.{CometBatchScanExec, CometNativeExec}
@@ -655,6 +659,122 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     global.result() ++ promoted.result()
   }
 
+  /**
+   * Declares the HDFS nameservices among `authorities` for iceberg-rust's HDFS backend. That
+   * backend dials an authority with a port as written, but resolves a portless one
+   * (`hdfs://ns1/...`) only through a nameservice declared in the FileIO properties, never
+   * through `$HADOOP_CONF_DIR`. A nameservice the Hadoop configuration declares in full, with
+   * `dfs.ha.namenodes.<ns>` and a `host:port` `dfs.namenode.rpc-address.<ns>.<id>` for each
+   * NameNode, becomes `hdfs.name-node.<ns>` with its NameNodes in declaration order. An authority
+   * the configuration knows nothing of is a single NameNode on Hadoop's default port, as
+   * `DFSUtilClient.getNNAddress` dials it for the JVM client. A name in `dfs.nameservices`
+   * without NameNodes, or with one that has no port, is left out, and
+   * [[undeclaredHdfsNameservice]] then declines the scan.
+   */
+  def hadoopToIcebergHdfsProperties(
+      hadoopConf: Configuration,
+      authorities: Set[String]): Map[String, String] = {
+    val logicalNames = hadoopConf.getTrimmedStringCollection("dfs.nameservices").asScala.toSet
+    hdfsNameservices(authorities).flatMap { ns =>
+      val nameNodes = hadoopConf
+        .getTrimmedStrings(s"dfs.ha.namenodes.$ns")
+        .map(id => hadoopConf.getTrimmed(s"dfs.namenode.rpc-address.$ns.$id"))
+      if (nameNodes.nonEmpty) {
+        if (nameNodes.forall(hasPort)) Some(s"hdfs.name-node.$ns" -> nameNodes.mkString(","))
+        else None
+      } else if (logicalNames.contains(ns)) {
+        None
+      } else {
+        Some(
+          s"hdfs.name-node.$ns" -> s"$ns:${HdfsClientConfigKeys.DFS_NAMENODE_RPC_PORT_DEFAULT}")
+      }
+    }.toMap
+  }
+
+  /**
+   * The first portless HDFS authority among `authorities` that `properties` declares neither with
+   * iceberg-rust's `hdfs.name-node.<ns>` nor with Hadoop's `dfs.ha.namenodes.<ns>`.
+   */
+  def undeclaredHdfsNameservice(
+      authorities: Set[String],
+      properties: Map[String, String]): Option[String] =
+    hdfsNameservices(authorities).find { ns =>
+      !properties.contains(s"hdfs.name-node.$ns") &&
+      !properties.contains(s"hadoop.dfs.ha.namenodes.$ns")
+    }
+
+  private def hdfsNameservices(authorities: Set[String]): Seq[String] =
+    authorities.filterNot(hasPort).toSeq.sorted
+
+  /**
+   * Encrypted flags of HDFS directories already probed, by filesystem URI and directory. A data
+   * directory cannot become a zone root once it holds files, so the answer is kept for the
+   * process; bounded so a long session over many tables stays small.
+   */
+  private val encryptedDirectories: java.util.Map[(String, String), java.lang.Boolean] =
+    java.util.Collections.synchronizedMap(
+      new java.util.LinkedHashMap[(String, String), java.lang.Boolean](64, 0.75f, true) {
+        override def removeEldestEntry(
+            eldest: java.util.Map.Entry[(String, String), java.lang.Boolean]): Boolean =
+          size() > 10000
+      })
+
+  /**
+   * The first of `directories`, HDFS directories of data and delete files, that lies in an HDFS
+   * encryption zone, or whose status could not be read. iceberg-rust's HDFS client is built
+   * without KMS support and rejects every file of a zone, so such a scan must not be claimed. The
+   * NameNode flags each path inside a zone as encrypted, directories included, so one listing of
+   * a parent tells the flag of every data directory under it: the partitions of a table cost one
+   * call, not one each, and a flag once read is cached. A cluster that cannot hold a zone at all,
+   * one whose NameNode advertises no key provider (a client-cached call) and whose session
+   * configures none, is not listed.
+   */
+  def hdfsEncryptedDirectory(
+      hadoopConf: Configuration,
+      directories: Set[String]): Option[String] = {
+    val configuredProvider =
+      Seq("hadoop.security.key.provider.path", "dfs.encryption.key.provider.uri")
+        .exists(key => hadoopConf.getTrimmed(key, "").nonEmpty)
+    // A directory at the root of the filesystem has no parent and is listed itself.
+    val byParent = directories.toSeq.sorted.map(new Path(_)).groupBy { dir =>
+      Option(dir.getParent).getOrElse(dir)
+    }
+    byParent.toSeq
+      .sortBy(_._1.toString)
+      .iterator
+      .map { case (parent, dirs) =>
+        try {
+          val fs = parent.getFileSystem(hadoopConf)
+          val key = (dir: Path) => (fs.getUri.toString, dir.toString)
+          val unknown = dirs.filterNot(dir => encryptedDirectories.containsKey(key(dir)))
+          val canHoldZones = unknown.nonEmpty && (configuredProvider ||
+            Option(fs.getServerDefaults(parent).getKeyProviderUri).exists(_.trim.nonEmpty))
+          if (canHoldZones) {
+            val listed =
+              fs.listStatus(parent).map(status => status.getPath.toString -> status.isEncrypted)
+            val flags = listed.toMap
+            unknown.foreach { dir =>
+              val flag = flags.getOrElse(dir.toString, fs.getFileStatus(dir).isEncrypted)
+              encryptedDirectories.put(key(dir), flag)
+            }
+          }
+          dirs.find(dir => encryptedDirectories.get(key(dir)) == java.lang.Boolean.TRUE)
+        } catch {
+          case NonFatal(e) =>
+            logWarning(s"Could not tell whether HDFS directories under $parent are encrypted", e)
+            dirs.headOption
+        }
+      }
+      .collectFirst { case Some(dir) => dir.toString }
+  }
+
+  private def hasPort(address: String): Boolean =
+    address != null && {
+      val colon = address.lastIndexOf(':')
+      colon > address.lastIndexOf(']') && colon < address.length - 1 &&
+      address.substring(colon + 1).forall(_.isDigit)
+    }
+
   /** iceberg-rust's per-I/O-call timeout, set from `spark.comet.iceberg.ioTimeout`. */
   def ioTimeoutProperty(): (String, String) =
     "opendal.io-timeout-ms" -> CometConf.COMET_ICEBERG_IO_TIMEOUT.get().toString
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergHdfsSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergHdfsSuite.scala
new file mode 100644
index 00000000000..dbe67d67037
--- /dev/null
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergHdfsSuite.scala
@@ -0,0 +1,497 @@
+/*
+ * 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.
+ */
+
+package org.apache.comet
+
+import java.io.File
+import java.net.URI
+import java.nio.file.Files
+import java.util.UUID
+
+import org.scalactic.source.Position
+import org.scalatest.Tag
+
+import org.apache.commons.io.FileUtils
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.crypto.key.{KeyProvider, KeyProviderFactory}
+import org.apache.hadoop.fs.{FileSystem, Path}
+import org.apache.hadoop.hdfs.{MiniDFSCluster, MiniDFSNNTopology}
+import org.apache.hadoop.hdfs.client.HdfsAdmin
+import org.apache.hadoop.util.VersionInfo
+import org.apache.iceberg.{DataFiles, FileFormat, Table}
+import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.comet.CometIcebergNativeScanExec
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+
+import org.apache.comet.CometSparkSessionExtensions.isSpark42Plus
+import org.apache.comet.iceberg.IcebergReflection
+
+/**
+ * End-to-end coverage of the native Iceberg scan against an `hdfs://` warehouse, backed by
+ * in-process `MiniDFSCluster`s.
+ *
+ * This is the only test that exercises iceberg-rust's `hdfs-native` backend for real. It matters
+ * because that backend is a second, independent HDFS client: the plain-Parquet native scan
+ * reaches HDFS through libhdfs/JNI (`fs.comet.libhdfs.schemes`), while an Iceberg table on HDFS
+ * is opened by a pure-Rust RPC client that shares nothing with it but the NameNode endpoints
+ * Comet hands it. A unit test over the scheme allowlists cannot tell whether that client actually
+ * connects and reads. Writes to HDFS use Iceberg's own writer and are not covered here.
+ *
+ * What is covered:
+ *   - scans of tables on a single-NameNode cluster, whose locations carry a real `host:port`
+ *     authority, so iceberg-rust needs no NameNode declaration;
+ *   - a two-NameNode HA cluster addressed by nameservice (`hdfs://<ns>/...`), which is not a
+ *     host, so the scan only connects through the `hdfs.name-node.<ns>` declaration Comet derives
+ *     from the session Hadoop configuration (iceberg-rust resolves a portless authority only
+ *     through it). It is read before and after a failover, and once the first NameNode is shut
+ *     down.
+ *
+ * Which profiles run it. The cluster needs the `hadoop-client-minicluster` that `pom.xml` pins
+ * per Spark profile (`hadoop.version`: 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) to match the `hadoop-client-api`/`runtime` Spark supplies. Measured by running
+ * the scan tests on every profile:
+ *   - Spark 3.4, 3.5, 4.0 and 4.1: the cluster starts and every test runs, none canceled. A
+ *     cluster that fails to start there is a broken fixture, so `beforeAll` fails the suite with
+ *     the cause instead of cancelling the tests.
+ *   - Spark 4.2: nothing runs (all tests are cancelled), for two independent reasons. There is no
+ *     Iceberg runtime for 4.2 (`icebergAvailable` is false there), and the cluster cannot start
+ *     either: `MiniDFSCluster.Builder` throws `NoClassDefFoundError:
+ *     org/junit/jupiter/api/Assertions`, because Hadoop 3.5.0's `GenericTestUtils` asserts with
+ *     JUnit 5, which the shaded minicluster leaves out and the test classpath does not carry.
+ *     With junit-jupiter-api 5.10.3 and its three small dependencies put on the JVM class path
+ *     the cluster starts, so a test-scoped `junit-jupiter-api` dependency on the 4.2 profile is
+ *     the fix once an Iceberg runtime exists. Until then this suite skips there and starts
+ *     nothing.
+ */
+class CometIcebergHdfsSuite
+    extends CometTestBase
+    with AdaptiveSparkPlanHelper
+    with CometIcebergTestBase
+    with WithHdfsCluster {
+
+  /**
+   * The only profile on which this suite does not run (see the class docstring): no Iceberg
+   * runtime, and no MiniDFSCluster. Anywhere else the Iceberg runtime and the cluster are
+   * required, and their absence fails the suite rather than cancelling it.
+   */
+  private def profileSkipped: Boolean = isSpark42Plus
+
+  override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
+      pos: Position): Unit = {
+    super.test(testName, testTags: _*) {
+      assume(!profileSkipped, "Spark 4.2 has no Iceberg runtime and no usable MiniDFSCluster")
+      testFun
+    }
+  }
+
+  /**
+   * The nameservice of the HA cluster. Not the `ns1` of `simpleHATopology`, to be unmistakable.
+   */
+  private val haNameservice = "cometha"
+
+  private var haCluster: MiniDFSCluster = _
+  private var haBaseDir: File = _
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    if (!profileSkipped) {
+      assert(icebergAvailable, "the Iceberg runtime is missing from the test classpath")
+      try startHdfsCluster()
+      catch {
+        case e @ (_: Exception | _: LinkageError) =>
+          throw new IllegalStateException(
+            "MiniDFSCluster failed to start, so none of the HDFS tests can run. " +
+              s"Spark ${org.apache.spark.SPARK_VERSION}, hadoop-client-api " +
+              s"${VersionInfo.getVersion}, hadoop-client-minicluster loaded from " +
+              s"${classOf[MiniDFSCluster].getProtectionDomain.getCodeSource.getLocation}: $e",
+            e)
+      }
+    }
+  }
+
+  override def afterAll(): Unit = {
+    try stopHaCluster()
+    finally {
+      try stopEncryptedCluster()
+      finally {
+        try stopHdfsCluster()
+        finally super.afterAll()
+      }
+    }
+  }
+
+  /** `hdfs://localhost:<port>` -- the authority iceberg-rust dials as the NameNode. */
+  private def hdfsUri: String = s"hdfs://localhost:$getDFSPort"
+
+  private def assertSingleNativeScan(cometPlan: SparkPlan): CometIcebergNativeScanExec = {
+    val scans = collect(cometPlan) { case scan: CometIcebergNativeScanExec => scan }
+    assert(
+      scans.length == 1,
+      s"Expected exactly 1 CometIcebergNativeScanExec but found ${scans.length}. " +
+        s"Plan:\n$cometPlan")
+    scans.head
+  }
+
+  /**
+   * Runs `f` with a Hadoop-catalog Iceberg warehouse rooted at `baseUri`, which is the
+   * single-NameNode cluster by default. The catalog name is unique per test so Spark's catalog
+   * cache cannot hand back a warehouse from an earlier test, and the warehouse directory is
+   * removed afterwards (`cleanup`).
+   */
+  private def withHdfsIcebergCatalog(baseUri: String = hdfsUri, cleanup: Boolean = true)(
+      f: String => Unit): Unit = {
+    val catalog = s"hdfs_cat_${UUID.randomUUID().toString.replace("-", "")}"
+    val warehouse = s"$baseUri/warehouse/${UUID.randomUUID()}"
+    try {
+      withSQLConf(
+        s"spark.sql.catalog.$catalog" -> "org.apache.iceberg.spark.SparkCatalog",
+        s"spark.sql.catalog.$catalog.type" -> "hadoop",
+        s"spark.sql.catalog.$catalog.warehouse" -> warehouse,
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+        f(catalog)
+      }
+    } finally {
+      if (cleanup) getFileSystem.delete(new Path(warehouse), true)
+    }
+  }
+
+  private def tableLocation(catalog: String, table: String): String = {
+    val icebergTable = loadIcebergTable(spark, catalog, "db", table)
+    icebergTable.getClass.getMethod("location").invoke(icebergTable).toString
+  }
+
+  test("native Iceberg scan reads a table stored on HDFS") {
+    withHdfsIcebergCatalog() { catalog =>
+      spark.sql(s"CREATE TABLE $catalog.db.t (id INT, name STRING, value DOUBLE) USING iceberg")
+      spark.sql(
+        s"INSERT INTO $catalog.db.t VALUES (1, 'Alice', 10.5), (2, 'Bob', 20.3), " +
+          "(3, 'Charlie', 30.7)")
+
+      // The data location must actually be on HDFS, or this suite would silently degrade into a
+      // duplicate of the local-filesystem coverage.
+      val dataLocation = IcebergReflection
+        .getDataLocation(loadIcebergTable(spark, catalog, "db", "t"))
+        .getOrElse(fail("could not resolve the Iceberg data location"))
+      assert(
+        dataLocation.startsWith("hdfs://"),
+        s"expected an hdfs:// data location, got $dataLocation")
+
+      val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.t ORDER BY id")
+      assertSingleNativeScan(cometPlan)
+
+      spark.sql(s"DROP TABLE $catalog.db.t")
+    }
+  }
+
+  test("native Iceberg scan on HDFS applies a pushed-down filter") {
+    withHdfsIcebergCatalog() { catalog =>
+      spark.sql(s"CREATE TABLE $catalog.db.f (id INT, name STRING) USING iceberg")
+      spark.sql(
+        s"INSERT INTO $catalog.db.f VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e')")
+
+      val (_, cometPlan) =
+        checkSparkAnswer(s"SELECT id, name FROM $catalog.db.f WHERE id > 3 ORDER BY id")
+      assertSingleNativeScan(cometPlan)
+
+      spark.sql(s"DROP TABLE $catalog.db.f")
+    }
+  }
+
+  test("native Iceberg scan reads a partitioned table across multiple HDFS data files") {
+    withHdfsIcebergCatalog() { catalog =>
+      spark.sql(
+        s"CREATE TABLE $catalog.db.p (id INT, part STRING) USING iceberg PARTITIONED BY (part)")
+      // Separate inserts so each partition lands in its own data file: a single-file read would
+      // not prove the operator cache serves more than one path from one NameNode.
+      spark.sql(s"INSERT INTO $catalog.db.p VALUES (1, 'x'), (2, 'x')")
+      spark.sql(s"INSERT INTO $catalog.db.p VALUES (3, 'y'), (4, 'y')")
+
+      val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.p ORDER BY id")
+      assertSingleNativeScan(cometPlan)
+
+      spark.sql(s"DROP TABLE $catalog.db.p")
+    }
+  }
+
+  private def stopHaCluster(): Unit = {
+    try if (haCluster != null) haCluster.shutdown(true)
+    finally {
+      haCluster = null
+      if (haBaseDir != null) FileUtils.deleteQuietly(haBaseDir)
+      haBaseDir = null
+    }
+  }
+
+  /**
+   * Starts a two-NameNode MiniDFSCluster with `nn1` active, and returns the client configuration
+   * a Spark job needs to address it by nameservice: exactly the keys a production `hdfs-site.xml`
+   * carries, so that Comet's derivation of `hdfs.name-node.<ns>` is what lets the native client
+   * dial.
+   */
+  private def startHaCluster(): Seq[(String, String)] = {
+    val conf = new Configuration()
+    conf.set("dfs.namenode.metrics.logger.period.seconds", "0")
+    conf.set("dfs.datanode.metrics.logger.period.seconds", "0")
+    conf.setIfUnset("dfs.namenode.rpc-bind-host", "localhost")
+    // A second cluster in this JVM must not format the first one's `target/test/data/dfs`.
+    haBaseDir = Files.createTempDirectory("comet_ha_dfs_").toFile
+    conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, haBaseDir.getAbsolutePath)
+
+    val topology = new MiniDFSNNTopology().addNameservice(
+      new MiniDFSNNTopology.NSConf(haNameservice)
+        .addNN(new MiniDFSNNTopology.NNConf("nn1"))
+        .addNN(new MiniDFSNNTopology.NNConf("nn2")))
+    haCluster = new MiniDFSCluster.Builder(conf).nnTopology(topology).numDataNodes(1).build()
+    haCluster.transitionToActive(0)
+
+    def rpcAddress(nn: String): String = {
+      val key = s"dfs.namenode.rpc-address.$haNameservice.$nn"
+      val address = haCluster.getConfiguration(0).get(key)
+      assert(address != null && address.nonEmpty, s"the HA cluster did not configure $key")
+      address
+    }
+    Seq(
+      "dfs.nameservices" -> haNameservice,
+      s"dfs.ha.namenodes.$haNameservice" -> "nn1,nn2",
+      s"dfs.namenode.rpc-address.$haNameservice.nn1" -> rpcAddress("nn1"),
+      s"dfs.namenode.rpc-address.$haNameservice.nn2" -> rpcAddress("nn2"),
+      s"dfs.client.failover.proxy.provider.$haNameservice" ->
+        "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider")
+  }
+
+  test("native Iceberg scan resolves an HA nameservice and follows a failover") {
+    val haClientConf = startHaCluster()
+    val logicalUri = new URI(s"hdfs://$haNameservice")
+    val haConf = haClientConf.toMap
+    // What Comet declares for the nameservice: its NameNodes in `dfs.ha.namenodes` order, as
+    // their rpc-addresses read.
+    val expectedNameNodes = Seq("nn1", "nn2")
+      .map(nn => haConf(s"dfs.namenode.rpc-address.$haNameservice.$nn"))
+      .mkString(",")
+
+    val hadoopConf = new Configuration()
+    haClientConf.foreach { case (k, v) => hadoopConf.set(k, v) }
+    try {
+      // The nameservice keys reach the session Hadoop configuration (planning, the Iceberg
+      // catalog, the commit) the way `spark.hadoop.*` settings would in a deployment.
+      withSQLConf(haClientConf: _*) {
+        withHdfsIcebergCatalog(baseUri = logicalUri.toString, cleanup = false) { catalog =>
+          // A location with no host:port, which iceberg-rust cannot dial by itself.
+          spark.sql(s"CREATE TABLE $catalog.db.ha (id INT, name STRING) USING iceberg")
+          spark.sql(s"INSERT INTO $catalog.db.ha VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+          val location = tableLocation(catalog, "ha")
+          assert(
+            location.startsWith(s"hdfs://$haNameservice/"),
+            s"expected a nameservice location, got $location")
+
+          // Succeeding is not enough: the declaration Comet derived is what the client was given.
+          // The per-nameservice key, since iceberg-rust reads the global `hdfs.name-node` only
+          // for authority-less paths.
+          def assertDeclared(properties: Map[String, String]): Unit = {
+            assert(
+              properties.get(s"hdfs.name-node.$haNameservice") == Some(expectedNameNodes),
+              s"$properties")
+            assert(!properties.contains("hdfs.name-node"), s"$properties")
+          }
+
+          def scanAndCheck(expectedRows: Long): Unit = {
+            val query = s"SELECT * FROM $catalog.db.ha ORDER BY id"
+            // Checked on the planned scan before anything runs, so a lost declaration fails here
+            // with the properties in the message, not as a NameNode resolution error on an
+            // executor after the native client's retries.
+            assertDeclared(
+              assertSingleNativeScan(
+                spark
+                  .sql(query)
+                  .queryExecution
+                  .executedPlan).nativeIcebergScanMetadata.catalogProperties)
+            val (_, cometPlan) = checkSparkAnswer(query)
+            assertDeclared(
+              assertSingleNativeScan(cometPlan).nativeIcebergScanMetadata.catalogProperties)
+            assert(
+              spark.sql(s"SELECT count(*) FROM $catalog.db.ha").collect().head.getLong(0) ==
+                expectedRows)
+          }
+
+          scanAndCheck(3L)
+
+          // Fail over: nn1 goes standby and nn2 takes over. The first NameNode on the native
+          // client's list is now a standby, and the list must carry it to the new active.
+          haCluster.transitionToStandby(0)
+          haCluster.transitionToActive(1)
+          assert(
+            haCluster.getNameNode(0).isStandbyState && haCluster.getNameNode(1).isActiveState)
+          scanAndCheck(3L)
+
+          // A NameNode that is gone rather than standby: connections to nn1 are now refused, and
+          // the cached native client must still reach nn2, for the files it knows and for one
+          // Iceberg's own writer adds through the HA client meanwhile.
+          haCluster.shutdownNameNode(0)
+          spark.sql(s"INSERT INTO $catalog.db.ha VALUES (4, 'd')")
+          scanAndCheck(4L)
+        }
+      }
+    } finally {
+      // Evict the cached client of the nameservice before its cluster goes away.
+      try FileSystem.get(logicalUri, hadoopConf).close()
+      catch { case _: Exception => () }
+      stopHaCluster()
+    }
+  }
+
+  private var ezCluster: MiniDFSCluster = _
+  private var ezBaseDir: File = _
+
+  private def stopEncryptedCluster(): Unit = {
+    try if (ezCluster != null) ezCluster.shutdown(true)
+    finally {
+      ezCluster = null
+      if (ezBaseDir != null) FileUtils.deleteQuietly(ezBaseDir)
+      ezBaseDir = null
+    }
+  }
+
+  /**
+   * Starts a cluster whose NameNode has a key provider (a `jceks` keystore on the local disk, as
+   * Hadoop's own encryption tests use), with one key and an encryption zone at `/ez`. The key is
+   * created before the NameNode starts, since it loads the keystore once.
+   */
+  private def startEncryptedCluster(): Unit = {
+    val conf = new Configuration()
+    conf.set("dfs.namenode.metrics.logger.period.seconds", "0")
+    conf.set("dfs.datanode.metrics.logger.period.seconds", "0")
+    conf.setIfUnset("dfs.namenode.rpc-bind-host", "localhost")
+    ezBaseDir = Files.createTempDirectory("comet_ez_dfs_").toFile
+    conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, ezBaseDir.getAbsolutePath)
+    val providerUri = s"jceks://file${new File(ezBaseDir, "ez.jceks").toURI.getPath}"
+    conf.set("hadoop.security.key.provider.path", providerUri)
+    val provider = KeyProviderFactory.get(new URI(providerUri), conf)
+    provider.createKey("ez-key", new KeyProvider.Options(conf))
+    provider.flush()
+
+    ezCluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build()
+    val fs = ezCluster.getFileSystem
+    fs.mkdirs(new Path("/ez"))
+    new HdfsAdmin(fs.getUri, ezCluster.getConfiguration(0))
+      .createEncryptionZone(new Path("/ez"), "ez-key")
+  }
+
+  test("native Iceberg scan falls back for a table in an HDFS encryption zone") {
+    startEncryptedCluster()
+    try {
+      val ezUri = s"hdfs://localhost:${ezCluster.getNameNodePort}"
+      // Iceberg's own writer encrypts through the key provider the NameNode advertises; the
+      // native reader cannot decrypt, so the scan must be declined at planning, not fail on an
+      // executor.
+      withHdfsIcebergCatalog(baseUri = s"$ezUri/ez", cleanup = false) { catalog =>
+        // Partitioned, so the probe meets several data directories under one parent.
+        spark.sql(
+          s"CREATE TABLE $catalog.db.enc (id INT, name STRING) USING iceberg " +
+            "PARTITIONED BY (name)")
+        spark.sql(s"INSERT INTO $catalog.db.enc VALUES (1, 'a'), (2, 'b')")
+        val dataLocation = IcebergReflection
+          .getDataLocation(loadIcebergTable(spark, catalog, "db", "enc"))
+          .getOrElse(fail("could not resolve the Iceberg data location"))
+        // The fixture really is a zone: the NameNode flags the data directory as encrypted.
+        assert(
+          ezCluster.getFileSystem.getFileStatus(new Path(dataLocation)).isEncrypted,
+          s"$dataLocation is not in the encryption zone")
+
+        val (_, cometPlan) = checkSparkAnswerAndFallbackReason(
+          s"SELECT * FROM $catalog.db.enc ORDER BY id",
+          "encryption zone")
+        val scans = collect(cometPlan) { case scan: CometIcebergNativeScanExec => scan }
+        assert(
+          scans.isEmpty,
+          s"an encryption-zone table must fall back, but was claimed:\n$cometPlan")
+      }
+      // The gate is per location, not per cluster: a table outside the zone on the same
+      // NameNode still scans natively.
+      withHdfsIcebergCatalog(baseUri = ezUri, cleanup = false) { catalog =>
+        spark.sql(s"CREATE TABLE $catalog.db.plain (id INT) USING iceberg")
+        spark.sql(s"INSERT INTO $catalog.db.plain VALUES (1), (2)")
+        val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.plain ORDER BY id")
+        assertSingleNativeScan(cometPlan)
+      }
+    } finally stopEncryptedCluster()
+  }
+
+  test("one native scan reads data files on two NameNodes") {
+    val haClientConf = startHaCluster()
+    val logicalUri = new URI(s"hdfs://$haNameservice")
+    val hadoopConf = new Configuration()
+    haClientConf.foreach { case (k, v) => hadoopConf.set(k, v) }
+    try {
+      withSQLConf(haClientConf: _*) {
+        // Metadata and the first data file on the single-NameNode cluster.
+        withHdfsIcebergCatalog() { catalog =>
+          spark.sql(s"CREATE TABLE $catalog.db.two (id INT, name STRING) USING iceberg")
+          spark.sql(s"INSERT INTO $catalog.db.two VALUES (1, 'a')")
+
+          // A second data file, written by Iceberg on the HA cluster and registered in the first
+          // table's manifest. Same DDL, so the field ids line up.
+          withHdfsIcebergCatalog(baseUri = logicalUri.toString, cleanup = false) { haCatalog =>
+            spark.sql(s"CREATE TABLE $haCatalog.db.src (id INT, name STRING) USING iceberg")
+            spark.sql(s"INSERT INTO $haCatalog.db.src VALUES (2, 'b')")
+            val src = loadIcebergTable(spark, haCatalog, "db", "src").asInstanceOf[Table]
+            val srcFile = src.newScan().planFiles().iterator().next().file()
+            val srcLocation = IcebergReflection
+              .extractFileLocation(
+                IcebergReflection.loadClass(IcebergReflection.ClassNames.CONTENT_FILE),
+                srcFile)
+              .getOrElse(fail("could not read the source data file location"))
+            assert(srcLocation.startsWith(s"hdfs://$haNameservice/"), srcLocation)
+
+            val table = loadIcebergTable(spark, catalog, "db", "two").asInstanceOf[Table]
+            table
+              .newAppend()
+              .appendFile(
+                DataFiles
+                  .builder(table.spec())
+                  .withPath(srcLocation)
+                  .withFormat(FileFormat.PARQUET)
+                  .withFileSizeInBytes(srcFile.fileSizeInBytes())
+                  .withRecordCount(srcFile.recordCount())
+                  .build())
+              .commit()
+            spark.sql(s"REFRESH TABLE $catalog.db.two")
+          }
+
+          // One FileIO serves both: the host:port authority dialed as written, the nameservice
+          // through the declaration.
+          val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.two ORDER BY id")
+          val scan = assertSingleNativeScan(cometPlan)
+          assert(
+            scan.nativeIcebergScanMetadata.catalogProperties
+              .contains(s"hdfs.name-node.$haNameservice"),
+            s"${scan.nativeIcebergScanMetadata.catalogProperties}")
+          assert(
+            spark.sql(s"SELECT count(*) FROM $catalog.db.two").collect().head.getLong(0) == 2L)
+        }
+      }
+    } finally {
+      try FileSystem.get(logicalUri, hadoopConf).close()
+      catch { case _: Exception => () }
+      stopHaCluster()
+    }
+  }
+}
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index 5dfbd19fcd0..d9d30cabb97 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -2833,11 +2833,12 @@ class CometIcebergNativeSuite
     }
   }
 
-  // Data files written without Iceberg field ids and read with no name mapping: iceberg-rust then
-  // projects whole top-level columns by position, and a by-name struct cast narrows each one to
-  // the pruned type. Spark's own reader returns NULL nested fields for such a table, so the pruned
-  // read is checked against the native read of every nested field instead.
-  test("nested schema pruning on data files without field ids") {
+  // Data files written without Iceberg field ids and read with no name mapping: neither reader
+  // can match such a file's nested fields to the table's, so a struct's fields read as NULL.
+  // Spark's reader gives them fallback readers; iceberg-rust, since apache/iceberg-rust#3437,
+  // treats them as absent, whether the struct is read whole or pruned. Top-level columns still
+  // read. A list of structs is the one shape the readers still disagree on.
+  test("nested fields of data files without field ids read as NULL, as in Spark") {
     assume(icebergAvailable, "Iceberg not available in classpath")
 
     withTempIcebergDir { warehouseDir =>
@@ -2897,24 +2898,35 @@ class CometIcebergNativeSuite
           spark.sql(
             s"ALTER TABLE $table UNSET TBLPROPERTIES IF EXISTS ('schema.name-mapping.default')")
 
-          def nativeRows(query: String, pruneNestedFields: Boolean): Seq[Row] = {
-            var rows = Seq.empty[Row]
+          // Both readers agree, with the struct read whole and read pruned.
+          Seq(
+            s"SELECT id, s FROM $table ORDER BY id",
+            s"SELECT id, s.b FROM $table ORDER BY id",
+            s"SELECT count(*) FROM $table WHERE s.a IS NULL",
+            s"SELECT count(*) FROM $table WHERE s IS NULL").foreach(checkPrunedAndFullReads)
+          // And NULL is what they agree on, not the file's values.
+          val rows = spark.sql(s"SELECT id, s.b FROM $table ORDER BY id").collect()
+          assert(rows.length == 100 && rows.forall(_.isNullAt(1)), rows.mkString(","))
+          // A list of structs: Spark's reader drops the elements and returns an empty list, the
+          // native read keeps each element with its fields NULL.
+          val listQuery = s"SELECT items.x FROM $table ORDER BY id"
+          Seq(true, false).foreach { pruneNestedFields =>
             withSQLConf(
               CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key ->
                 pruneNestedFields.toString) {
-              val df = spark.sql(query)
-              rows = df.collect().toSeq
+              val df = spark.sql(listQuery)
+              val native = df.collect()
               assertSingleNativeScan(df.queryExecution.executedPlan)
+              assert(
+                native.length == 100 && native.forall(_.getSeq[Any](0) == Seq(null)),
+                native.take(3).mkString(","))
             }
-            rows
           }
-          // `s.b` is the struct's last field, so a positional cast would read `s.a` instead.
-          Seq(
-            s"SELECT id, s.b FROM $table ORDER BY id",
-            s"SELECT id, items.x FROM $table ORDER BY id").foreach { query =>
-            val full = nativeRows(query, pruneNestedFields = false)
-            assert(full.length == 100 && full.forall(!_.isNullAt(1)), s"$query: $full")
-            assert(nativeRows(query, pruneNestedFields = true) == full, query)
+          withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+            val vanilla = spark.sql(listQuery).collect()
+            assert(
+              vanilla.length == 100 && vanilla.forall(_.getSeq[Any](0).isEmpty),
+              vanilla.take(3).mkString(","))
           }
 
           spark.sql(s"DROP TABLE $table")
@@ -2927,6 +2939,102 @@ class CometIcebergNativeSuite
     }
   }
 
+  // A table migrated from Spark-written Parquet reads its files, which have no field ids, through
+  // its name mapping. iceberg-rust cannot open the list layout that `spark.sql.parquet.writeLegacyFormat`
+  // writes (a two-level list) through a name mapping: the read schema it supplies keeps a field
+  // id on the list's element field and the Parquet reader rejects it. The layout is only known
+  // from the file, so the scan falls back for the list column and stays native for the primitives.
+  test("name-mapped table with a list column from a legacy Parquet layout falls back") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+
+    withTempIcebergDir { warehouseDir =>
+      withSQLConf(
+        "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+        "spark.sql.catalog.test_cat.type" -> "hadoop",
+        "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+        val dataPath = s"${warehouseDir.getAbsolutePath}/legacy_list"
+        // Non-null elements and a non-null struct field: with `writeLegacyFormat` Spark then writes
+        // the two-level layout, `repeated group array { required int32 col2 }`, the list's
+        // element being the repeated group itself.
+        val sparkSchema = StructType(
+          Seq(
+            org.apache.spark.sql.types.StructField("id", IntegerType, nullable = false),
+            org.apache.spark.sql.types.StructField(
+              "col1",
+              ArrayType(
+                StructType(Seq(
+                  org.apache.spark.sql.types.StructField("col2", IntegerType, nullable = false))),
+                containsNull = false))))
+        val jsonRows = (0 until 10).map(i => s"""{"id": $i, "col1": [{"col2": ${i * 2}}]}""")
+        withSQLConf("spark.sql.parquet.writeLegacyFormat" -> "true") {
+          spark.read
+            .schema(sparkSchema)
+            .json(spark.createDataset(jsonRows)(org.apache.spark.sql.Encoders.STRING))
+            .coalesce(1)
+            .write
+            .parquet(dataPath)
+        }
+        spark.sql("CREATE NAMESPACE IF NOT EXISTS test_cat.db")
+        val table = "test_cat.db.legacy_list"
+        spark.sql(s"CREATE TABLE $table (id INT, col1 ARRAY<STRUCT<col2: INT>>) USING iceberg")
+        try {
+          val tableUtilClass = Class.forName("org.apache.iceberg.spark.SparkTableUtil")
+          val icebergTable = spark.sessionState.catalogManager
+            .catalog("test_cat")
+            .asInstanceOf[org.apache.iceberg.spark.SparkCatalog]
+            .loadTable(
+              org.apache.spark.sql.connector.catalog.Identifier.of(Array("db"), "legacy_list"))
+            .asInstanceOf[org.apache.iceberg.spark.source.SparkTable]
+            .table()
+          // What the migrate and snapshot actions set for a table they create: `importSparkTable`
+          // reads a mapping that is already there and does not create one.
+          icebergTable
+            .updateProperties()
+            .set(
+              "schema.name-mapping.default",
+              org.apache.iceberg.mapping.NameMappingParser
+                .toJson(org.apache.iceberg.mapping.MappingUtil.create(icebergTable.schema())))
+            .commit()
+          spark.sql(s"CREATE TABLE parquet_legacy_list USING parquet LOCATION '$dataPath'")
+          tableUtilClass
+            .getMethod(
+              "importSparkTable",
+              classOf[org.apache.spark.sql.SparkSession],
+              classOf[org.apache.spark.sql.catalyst.TableIdentifier],
+              classOf[org.apache.iceberg.Table],
+              classOf[String])
+            .invoke(
+              null,
+              spark,
+              new org.apache.spark.sql.catalyst.TableIdentifier("parquet_legacy_list"),
+              icebergTable,
+              s"${warehouseDir.getAbsolutePath}/staging")
+
+          // The list column: declined at planning, and Spark reads the file's values.
+          val (_, listPlan) = checkSparkAnswerAndFallbackReason(
+            s"SELECT id, col1 FROM $table ORDER BY id",
+            "Nested columns with Iceberg name mapping")
+          assert(
+            collect(listPlan) { case s: CometIcebergNativeScanExec => s }.isEmpty,
+            s"$listPlan")
+          val read = spark.sql(s"SELECT col1[0].col2 FROM $table ORDER BY id").collect()
+          assert(read.map(_.getInt(0)).toSeq == (0 until 10).map(_ * 2), read.mkString(","))
+
+          // The primitives alone stay native.
+          val (_, idPlan) = checkSparkAnswer(s"SELECT id FROM $table ORDER BY id")
+          assertSingleNativeScan(idPlan)
+        } finally {
+          spark.sql(s"DROP TABLE IF EXISTS $table")
+          spark.sql("DROP TABLE IF EXISTS parquet_legacy_list")
+        }
+      }
+    }
+  }
+
   test("UUID type - native Iceberg UUID column (reproduces type mismatch)") {
     assume(icebergAvailable, "Iceberg not available in classpath")
 
@@ -4521,27 +4629,31 @@ class CometIcebergNativeSuite
             distinctCount == numRows,
             s"Expected $numRows distinct IDs but got $distinctCount")
 
-          // Spark's Iceberg reader returns null for INT96 timestamps inside structs,
-          // so we can't use checkIcebergNativeScan (which compares against Spark) for
-          // ts_struct. Instead, compare Comet's read against the raw Parquet source.
           checkIcebergNativeScan(
             "SELECT id, ts, value, ts_array, ts_map FROM test_cat.db.int96_test ORDER BY id")
           checkIcebergNativeScan("SELECT id, ts FROM test_cat.db.int96_test ORDER BY id")
           checkIcebergNativeScan("SELECT id, ts_array FROM test_cat.db.int96_test ORDER BY id")
           checkIcebergNativeScan("SELECT id, ts_map FROM test_cat.db.int96_test ORDER BY id")
-
-          // Validate ts_struct against raw Parquet since Spark's Iceberg reader can't read it
-          val icebergStructDf = spark
-            .sql("SELECT id, ts_struct FROM test_cat.db.int96_test ORDER BY id")
-            .collect()
-          val parquetStructDf = spark.read
-            .parquet(dataPath)
-            .select("id", "ts_struct")
-            .orderBy("id")
-            .collect()
+          // The imported file has no field ids, so neither reader can read the struct's fields
+          // (apache/iceberg-rust#3437): the native scan returns the struct with NULL fields, and
+          // Spark's reader, which cannot tell an optional struct's own nullness without a field
+          // to read, returns NULL for the struct itself. The native read still decodes the
+          // INT96 leaf, since such a file is read by whole columns, so an overflow would
+          // surface here.
+          val structQuery = "SELECT id, ts_struct FROM test_cat.db.int96_test ORDER BY id"
+          val structDf = spark.sql(structQuery)
+          val nativeStructs = structDf.collect()
+          assertSingleNativeScan(structDf.queryExecution.executedPlan)
           assert(
-            icebergStructDf.sameElements(parquetStructDf),
-            "ts_struct mismatch between Comet Iceberg read and raw Parquet")
+            nativeStructs.length == numRows &&
+              nativeStructs.forall(_.getStruct(1) == Row(null, null)),
+            nativeStructs.take(3).mkString(","))
+          withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+            val sparkStructs = spark.sql(structQuery).collect()
+            assert(
+              sparkStructs.length == numRows && sparkStructs.forall(_.isNullAt(1)),
+              sparkStructs.take(3).mkString(","))
+          }
 
           spark.sql("DROP TABLE test_cat.db.int96_test")
           spark.sql("DROP TABLE parquet_temp")
diff --git a/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala
index 8cb1765fa39..22a5486a0a4 100644
--- a/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala
@@ -28,16 +28,18 @@ import java.util.concurrent.atomic.AtomicLong
 import scala.util.Random
 
 import org.apache.commons.io.FileUtils
+import org.apache.iceberg.{DataFiles, FileFormat, Table}
 import org.apache.spark.SparkConf
 import org.apache.spark.paths.SparkPath
 import org.apache.spark.sql.{CometTestBase, SaveMode}
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.comet.{CometIcebergNativeScanExec, CometIcebergWriteExec, CometScanExec, CometScanUtils}
+import org.apache.spark.sql.comet.{CometBatchScanExec, CometIcebergNativeScanExec, CometIcebergWriteExec, CometScanExec, CometScanUtils}
 import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
 import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 import org.apache.spark.sql.internal.SQLConf
 
-import org.apache.comet.{CometConf, CometIcebergTestBase, ExtendedExplainInfo, NativeBase}
+import org.apache.comet.{CometConf, CometExplainInfo, CometIcebergTestBase, ExtendedExplainInfo, NativeBase}
 import org.apache.comet.hadoop.fs.{FakeHdfsAuthorityFileSystem, FakeHDFSFileSystem, FakeHdfsSchemeFileSystem, FakeWasbSchemeFileSystem}
 import org.apache.comet.iceberg.IcebergStorageSchemes
 import org.apache.comet.objectstore.NativeConfig
@@ -154,6 +156,10 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa
     assert(
       !write.contains("oss"),
       s"oss must not be in the native write schemes ${write.toSeq.sorted}")
+    assert(read.contains("hdfs"), s"hdfs must be in the native read schemes ${read.toSeq.sorted}")
+    assert(
+      !write.contains("hdfs"),
+      s"hdfs must not be in the native write schemes ${write.toSeq.sorted}")
     assert(
       write.contains("memory"),
       s"memory must be in the native write schemes ${write.toSeq.sorted}")
@@ -181,7 +187,8 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa
       "s3://bucket/key.parquet",
       "s3a://bucket/key.parquet",
       "gs://bucket/key.parquet",
-      "oss://bucket/key.parquet").foreach { u =>
+      "oss://bucket/key.parquet",
+      "hdfs://nn:8020/key.parquet").foreach { u =>
       assert(
         CometScanRule.isIcebergReadableScheme(new URI(u), Set.empty),
         s"$u must be iceberg-readable; icebergReadableSchemes has regressed")
@@ -284,6 +291,9 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa
     assert(
       !openable("s3:///bucket/k.parquet"),
       "authorityless s3:/// has no host and no alias promotion")
+    assert(
+      !openable("hdfs:///warehouse/k.parquet"),
+      "authorityless hdfs:/// needs a default NameNode, which Comet does not pass to native")
     // blob not opted in: no promotion, so the hostless form is not openable either.
     assert(
       !openable("blob:///bucket/k.parquet", Set.empty),
@@ -740,6 +750,278 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa
         2 -> Seq("s3a://b/4.csv")))
     assert(split.flatMap(_.files.map(_.partitionValues.getInt(0))) == Seq(1, 3, 2, 4))
   }
+
+  // --- Iceberg scans over HDFS data files ---------------------------------------------------
+  //
+  // Planner-level: a real CometScanRule over a real Iceberg table whose data files are registered
+  // in the manifests with the `hdfs://` paths the test names. Planning reads manifests, never the
+  // data files, so nothing here needs a NameNode and nothing is executed.
+
+  /**
+   * Plans `SELECT id` over an Iceberg table whose data files sit at `dataFiles`, applies
+   * `CometScanRule` with native Iceberg scans enabled and `hadoopConf` set on the session, and
+   * returns the scans the rule claimed plus the fallback reasons it left on the ones it declined.
+   * `catalogProps` are set on the catalog before its first use, so its FileIO is built with them.
+   * The warehouse, and so the table metadata, sits on the fake `hdfs://` filesystem
+   * (`metadataPrefix`): the native reader opens every file through one storage chosen from the
+   * metadata location, so `hdfs://` data files are claimed only under `hdfs://` metadata. An
+   * empty prefix leaves the metadata on the local filesystem.
+   */
+  private def planIcebergScanOverHdfs(
+      dataFiles: Seq[String],
+      hadoopConf: Seq[(String, String)] = Seq.empty,
+      catalogProps: Seq[(String, String)] = Seq.empty,
+      metadataPrefix: String = FakeHdfsSchemeFileSystem.PREFIX)
+      : (Seq[CometBatchScanExec], Set[String]) = {
+    // A catalog of its own, so no test reads another's cached catalog or warehouse.
+    val catalog = s"hdfs_gate_${UUID.randomUUID().toString.take(8)}"
+    var claimed = Seq.empty[CometBatchScanExec]
+    var reasons = Set.empty[String]
+    val catalogConfs = catalogProps.map { case (k, v) => s"spark.sql.catalog.$catalog.$k" -> v }
+    withTempIcebergDir { dir =>
+      val confs = Seq(
+        s"spark.sql.catalog.$catalog" -> "org.apache.iceberg.spark.SparkCatalog",
+        s"spark.sql.catalog.$catalog.type" -> "hadoop",
+        s"spark.sql.catalog.$catalog.warehouse" ->
+          s"$metadataPrefix${dir.getAbsolutePath}/warehouse",
+        // Iceberg fetches block locations of `hdfs://` data files when the table is on HDFS;
+        // the files here exist only in the manifests.
+        "spark.sql.iceberg.locality.enabled" -> "false") ++ catalogConfs
+      withSQLConf(confs: _*) {
+        spark.sql(s"CREATE TABLE $catalog.db.t (id INT) USING iceberg")
+        val table = loadIcebergTable(spark, catalog, "db", "t").asInstanceOf[Table]
+        val append = table.newAppend()
+        dataFiles.foreach { path =>
+          append.appendFile(
+            DataFiles
+              .builder(table.spec())
+              .withPath(path)
+              .withFormat(FileFormat.PARQUET)
+              .withFileSizeInBytes(1024)
+              .withRecordCount(1)
+              .build())
+        }
+        append.commit()
+
+        // The Spark plan first, with Comet off, then the rule alone: no execution.
+        var sparkPlan: SparkPlan = null
+        withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+          sparkPlan = spark.sql(s"SELECT id FROM $catalog.db.t").queryExecution.executedPlan
+        }
+        val confs = Seq(
+          CometConf.COMET_ENABLED.key -> "true",
+          CometConf.COMET_EXEC_ENABLED.key -> "true",
+          CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") ++ hadoopConf
+        withSQLConf(confs: _*) {
+          val transformed = CometScanRule(spark).apply(stripAQEPlan(sparkPlan))
+          claimed = transformed.collect { case s: CometBatchScanExec => s }
+          reasons = transformed
+            .collect { case s: BatchScanExec =>
+              s.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String])
+            }
+            .flatten
+            .toSet
+        }
+      }
+    }
+    (claimed, reasons)
+  }
+
+  private def nameNodeProperties(scan: CometBatchScanExec): Map[String, String] =
+    scan.nativeIcebergScanMetadata.get.catalogProperties.filter { case (k, _) =>
+      k.startsWith("hdfs.name-node")
+    }
+
+  private val declaredNs1 = Seq(
+    "dfs.nameservices" -> "ns1",
+    "dfs.ha.namenodes.ns1" -> "nn1,nn2",
+    "dfs.namenode.rpc-address.ns1.nn1" -> "nn1.example.com:9000",
+    "dfs.namenode.rpc-address.ns1.nn2" -> "nn2.example.com:8020")
+
+  private val undeclaredReason = "which has no port and which the Hadoop configuration does not"
+
+  private val mixedStorageReason = "through one storage backend"
+
+  test("iceberg scan: an HDFS nameservice the Hadoop configuration declares is claimed") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    val (claimed, reasons) =
+      planIcebergScanOverHdfs(Seq("hdfs://ns1/warehouse/db/t/a.parquet"), declaredNs1)
+    assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons")
+    // The fixture must really put the metadata on hdfs://, or these tests would exercise the
+    // mixed-storage gate instead of the nameservice declaration.
+    assert(
+      claimed.head.nativeIcebergScanMetadata.get.metadataLocation
+        .startsWith(FakeHdfsSchemeFileSystem.PREFIX),
+      s"metadata location: ${claimed.head.nativeIcebergScanMetadata.get.metadataLocation}")
+    // The per-nameservice key, with the NameNodes in `dfs.ha.namenodes` order; iceberg-rust
+    // reads the global `hdfs.name-node` only for authority-less paths.
+    assert(
+      nameNodeProperties(claimed.head) ==
+        Map("hdfs.name-node.ns1" -> "nn1.example.com:9000,nn2.example.com:8020"),
+      s"unexpected properties: ${claimed.head.nativeIcebergScanMetadata.get.catalogProperties}")
+  }
+
+  test("iceberg scan: a nameservice with a NameNode missing its rpc-address is declined") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // nn2 has no rpc-address, so the declaration is all-or-nothing empty and the executor would
+    // be left to resolve `ns1` as if it were a host.
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://ns1/warehouse/db/t/a.parquet"),
+      declaredNs1.filterNot(_._1 == "dfs.namenode.rpc-address.ns1.nn2"))
+    assert(claimed.isEmpty, "an undeclared nameservice must not be claimed natively")
+    assert(
+      reasons.exists(r => r.contains("HDFS authority 'ns1'") && r.contains(undeclaredReason)),
+      s"expected the nameservice fall-back reason, got: $reasons")
+  }
+
+  test("iceberg scan: a nameservice listed only in dfs.nameservices is declined") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://ns1/warehouse/db/t/a.parquet"),
+      Seq("dfs.nameservices" -> "ns1,ns2"))
+    assert(claimed.isEmpty, "a nameservice without NameNodes must not be claimed")
+    assert(
+      reasons.exists(r => r.contains("HDFS authority 'ns1'") && r.contains(undeclaredReason)),
+      s"expected the nameservice fall-back reason, got: $reasons")
+  }
+
+  test("iceberg scan: a portless plain HDFS host is claimed on Hadoop's default port") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // iceberg-rust has no default NameNode port, so `hdfs://nn.example.com/...` would be a
+    // nameservice name to it; Comet declares it on 8020, as the JVM client dials it.
+    val (claimed, reasons) =
+      planIcebergScanOverHdfs(Seq("hdfs://nn.example.com/warehouse/db/t/a.parquet"))
+    assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons")
+    assert(
+      nameNodeProperties(claimed.head) ==
+        Map("hdfs.name-node.nn.example.com" -> "nn.example.com:8020"))
+  }
+
+  test("iceberg scan: HDFS files are declined when executors would read as another user") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // The native client connects as HADOOP_USER_NAME or the executor's OS user, never as the
+    // Spark user the JVM reader uses; the scan is claimed only when those are known to agree.
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"),
+      Seq("spark.executorEnv.HADOOP_USER_NAME" -> "someone-else"))
+    assert(claimed.isEmpty, "a scan read as another user must not be claimed")
+    assert(
+      reasons.exists(r =>
+        r.contains("'someone-else'") && r.contains("spark.executorEnv.HADOOP_USER_NAME")),
+      s"expected the identity fall-back reason, got: $reasons")
+  }
+
+  test("iceberg scan: HDFS files are claimed when HADOOP_USER_NAME is the Spark user") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"),
+      Seq("spark.executorEnv.HADOOP_USER_NAME" -> spark.sparkContext.sparkUser))
+    assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons")
+  }
+
+  test("iceberg scan: a host:port HDFS location is claimed with no NameNode declaration") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    val (claimed, reasons) =
+      planIcebergScanOverHdfs(Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"))
+    assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons")
+    assert(nameNodeProperties(claimed.head).isEmpty)
+  }
+
+  test("iceberg scan: data files on two declared nameservices are claimed, each declared") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // iceberg-rust resolves each path's authority on its own and keeps one client per NameNode,
+    // so a scan may span clusters as long as every nameservice is declared.
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://ns1/warehouse/db/t/a.parquet", "hdfs://ns2/warehouse/db/t/b.parquet"),
+      declaredNs1 ++ Seq(
+        "dfs.ha.namenodes.ns2" -> "nn1",
+        "dfs.namenode.rpc-address.ns2.nn1" -> "nn3.example.com:8020"))
+    assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons")
+    assert(
+      nameNodeProperties(claimed.head) == Map(
+        "hdfs.name-node.ns1" -> "nn1.example.com:9000,nn2.example.com:8020",
+        "hdfs.name-node.ns2" -> "nn3.example.com:8020"))
+  }
+
+  test("iceberg scan: a catalog hdfs.name-node.<ns> wins over the Hadoop-derived declaration") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://ns1/warehouse/db/t/a.parquet"),
+      declaredNs1,
+      Seq("hdfs.name-node.ns1" -> "nn9.example.com:8020"))
+    assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons")
+    assert(
+      nameNodeProperties(claimed.head) == Map("hdfs.name-node.ns1" -> "nn9.example.com:8020"),
+      "the catalog value must reach native")
+  }
+
+  test("iceberg scan: a nameservice declared only by the catalog's hadoop.* keys is claimed") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // Comet derives declarations from the session configuration alone, but iceberg-rust reads
+    // Hadoop's own keys from the properties too, which a catalog forwards as `hadoop.*`.
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://nsb/warehouse/db/t/a.parquet"),
+      catalogProps = Seq(
+        "hadoop.dfs.nameservices" -> "nsb",
+        "hadoop.dfs.ha.namenodes.nsb" -> "nn1,nn2",
+        "hadoop.dfs.namenode.rpc-address.nsb.nn1" -> "a.example.com:8020",
+        "hadoop.dfs.namenode.rpc-address.nsb.nn2" -> "b.example.com:8020"))
+    assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons")
+    val properties = claimed.head.nativeIcebergScanMetadata.get.catalogProperties
+    assert(properties.get("hadoop.dfs.ha.namenodes.nsb") == Some("nn1,nn2"), s"$properties")
+    assert(!properties.contains("hdfs.name-node.nsb"), s"$properties")
+  }
+
+  test("iceberg scan: HDFS data files of a table whose metadata is local are declined") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // The native scan builds its one storage from the metadata location, a local path here, and
+    // would open the `hdfs://` file through it; declined at planning rather than failing at open.
+    val (claimed, reasons) = planIcebergScanOverHdfs(
+      Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"),
+      metadataPrefix = "")
+    assert(claimed.isEmpty, "a local-metadata table with HDFS data files must not be claimed")
+    assert(
+      reasons.exists(r => r.contains("'hdfs'") && r.contains(mixedStorageReason)),
+      s"expected the mixed-storage fall-back reason, got: $reasons")
+  }
+
+  test("iceberg scan: a local data file of a table whose metadata is on HDFS is declined") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    val (claimed, reasons) =
+      planIcebergScanOverHdfs(Seq(s"file://${fakeRootDir.getAbsolutePath}/t/a.parquet"))
+    assert(claimed.isEmpty, "an HDFS-metadata table with a local data file must not be claimed")
+    assert(
+      reasons.exists(r => r.contains("'file'") && r.contains(mixedStorageReason)),
+      s"expected the mixed-storage fall-back reason, got: $reasons")
+  }
+
+  test("iceberg scan: S3 data files of a table whose metadata is local are declined") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // Not an HDFS matter: the single storage fails the same way for an object store.
+    val (claimed, reasons) =
+      planIcebergScanOverHdfs(Seq("s3://bucket/warehouse/db/t/a.parquet"), metadataPrefix = "")
+    assert(claimed.isEmpty, "a local-metadata table with S3 data files must not be claimed")
+    assert(
+      reasons.exists(r => r.contains("'s3'") && r.contains(mixedStorageReason)),
+      s"expected the mixed-storage fall-back reason, got: $reasons")
+  }
+
+  test("iceberg scan: HDFS data files on a Kerberos cluster are declined") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    // The native client cannot use an executor's delegation tokens, so any HDFS location falls
+    // back once the cluster is secured, declared nameservice or not.
+    for (location <- Seq(
+        "hdfs://ns1/warehouse/db/t/a.parquet",
+        "hdfs://nn.example.com:8020/warehouse/db/t/a.parquet")) {
+      val (claimed, reasons) = planIcebergScanOverHdfs(
+        Seq(location),
+        declaredNs1 :+ ("hadoop.security.authentication" -> "kerberos"))
+      assert(claimed.isEmpty, s"$location must not be claimed on a Kerberos cluster")
+      assert(
+        reasons.exists(_.contains("cannot authenticate to HDFS with delegation tokens")),
+        s"expected the Kerberos fall-back reason for $location, got: $reasons")
+    }
+  }
 }
 
 /** A store key that counts the calls to its `equals`. */
diff --git a/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala b/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala
index 6ff5daf27fe..3408f5c9924 100644
--- a/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala
@@ -19,11 +19,17 @@
 
 package org.apache.comet.serde.operator
 
+import java.io.File
 import java.lang.reflect.InvocationTargetException
+import java.net.URI
+import java.nio.file.Files
+import java.util.concurrent.atomic.AtomicInteger
 
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
 
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, Path, RawLocalFileSystem}
 import org.apache.iceberg.{DeleteFile, FileContent}
 import org.apache.iceberg.expressions.Expressions
 import org.apache.spark.sql.catalyst.expressions.AttributeReference
@@ -240,6 +246,90 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers {
     out.values.toSet should not contain "https://some.example.com"
   }
 
+  test("HDFS nameservices are declared from the Hadoop configuration") {
+    val conf = new Configuration(false)
+    conf.set("dfs.ha.namenodes.ns1", "nn2, nn1")
+    conf.set("dfs.namenode.rpc-address.ns1.nn1", "host1:8020")
+    conf.set("dfs.namenode.rpc-address.ns1.nn2", "host2:8020")
+    // A NameNode without a port cannot be dialed, so ns2 is not declared.
+    conf.set("dfs.ha.namenodes.ns2", "nn1")
+    conf.set("dfs.namenode.rpc-address.ns2.nn1", "host3")
+    val authorities = Set("ns1", "ns2", "ns3", "host4:8020")
+
+    val properties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, authorities)
+
+    // NameNodes keep their declaration order, an authority with a port needs no declaration, and
+    // one the configuration knows nothing of is a host on the default port.
+    properties shouldBe Map(
+      "hdfs.name-node.ns1" -> "host2:8020,host1:8020",
+      "hdfs.name-node.ns3" -> "ns3:8020")
+    CometIcebergNativeScan.undeclaredHdfsNameservice(authorities, properties) shouldBe Some("ns2")
+    // A catalog's own declaration counts in either form.
+    val catalog = Map("hdfs.name-node.ns2" -> "host3:8020", "hadoop.dfs.ha.namenodes.ns3" -> "a")
+    CometIcebergNativeScan.undeclaredHdfsNameservice(
+      authorities,
+      properties ++ catalog) shouldBe None
+  }
+
+  test("HDFS authorities with a port need no declaration, IPv6 literals included") {
+    val conf = new Configuration(false)
+    conf.set("dfs.ha.namenodes.ns1", "nn1")
+    conf.set("dfs.namenode.rpc-address.ns1.nn1", "[::1]:8020")
+    val authorities = Set("[::1]:8020", "[2001:db8::1]:9000", "[::1]", "ns1")
+
+    val properties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, authorities)
+
+    // The colons inside the brackets are not a port separator: a bracketed rpc-address keeps
+    // its port, and a bracketed literal without one is a portless host, so it gets the default
+    // port like any other undeclared host.
+    properties shouldBe Map(
+      "hdfs.name-node.ns1" -> "[::1]:8020",
+      "hdfs.name-node.[::1]" -> "[::1]:8020")
+    CometIcebergNativeScan.undeclaredHdfsNameservice(authorities, properties) shouldBe None
+  }
+
+  test("an undeclared portless HDFS host is declared on Hadoop's default NameNode port") {
+    val conf = new Configuration(false)
+    conf.set("dfs.nameservices", "ns1")
+    val authorities = Set("nn.example.com", "ns1")
+
+    val properties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, authorities)
+
+    // What DFSUtilClient.getNNAddress does for the JVM client: a host without a port is dialed on
+    // 8020. A name in dfs.nameservices is logical to Hadoop too; without NameNodes it stays
+    // undeclared rather than becoming a host.
+    properties shouldBe Map("hdfs.name-node.nn.example.com" -> "nn.example.com:8020")
+    CometIcebergNativeScan.undeclaredHdfsNameservice(authorities, properties) shouldBe Some("ns1")
+  }
+
+  test("the encryption-zone probe lists each parent directory once and caches the answer") {
+    val root = Files.createTempDirectory("comet_ez_probe").toFile
+    val dirs = (1 to 3).map { i =>
+      val dir = new File(root, s"data/p=$i")
+      assert(dir.mkdirs())
+      s"${CountingHdfsFileSystem.PREFIX}${dir.getAbsolutePath}"
+    }.toSet
+    val conf = new Configuration(false)
+    conf.set("fs.hdfs.impl", classOf[CountingHdfsFileSystem].getName)
+    conf.set("fs.hdfs.impl.disable.cache", "true")
+    // A configured key provider makes the probe run; the local filesystem has no zones.
+    conf.set("hadoop.security.key.provider.path", "kms://http@kms.example.com:9600/kms")
+    CountingHdfsFileSystem.reset()
+
+    CometIcebergNativeScan.hdfsEncryptedDirectory(conf, dirs) shouldBe None
+    // The three partition directories share a parent: one listing, no per-directory status.
+    CountingHdfsFileSystem.listStatusCalls.get() shouldBe 1
+    CountingHdfsFileSystem.getFileStatusCalls.get() shouldBe 0
+
+    CometIcebergNativeScan.hdfsEncryptedDirectory(conf, dirs) shouldBe None
+    CountingHdfsFileSystem.listStatusCalls.get() shouldBe 1
+
+    // A directory at the filesystem root has no parent to list.
+    CometIcebergNativeScan.hdfsEncryptedDirectory(
+      conf,
+      Set(s"${CountingHdfsFileSystem.PREFIX}/")) shouldBe None
+  }
+
   private def keyMetadataMethod(clazz: Class[_]) = clazz.getMethod("keyMetadata")
 
   private def serializeDeleteFile(file: AnyRef) =
@@ -395,3 +485,35 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers {
     def keyMetadata(): java.nio.ByteBuffer = null
   }
 }
+
+/**
+ * A local-disk `hdfs://` filesystem that counts the NameNode calls the planner would make. The
+ * local `listStatus` stats each child itself, which a NameNode listing does not, so those inner
+ * calls are not counted.
+ */
+class CountingHdfsFileSystem extends RawLocalFileSystem {
+  RawLocalFileSystem.useStatIfAvailable()
+  private var listing = false
+  override def getScheme: String = "hdfs"
+  override def getUri: URI = URI.create(CountingHdfsFileSystem.PREFIX)
+  override def listStatus(path: Path): Array[FileStatus] = {
+    CountingHdfsFileSystem.listStatusCalls.incrementAndGet()
+    listing = true
+    try super.listStatus(path)
+    finally listing = false
+  }
+  override def getFileStatus(path: Path): FileStatus = {
+    if (!listing) CountingHdfsFileSystem.getFileStatusCalls.incrementAndGet()
+    super.getFileStatus(path)
+  }
+}
+
+object CountingHdfsFileSystem {
+  val PREFIX = "hdfs://counting"
+  val listStatusCalls = new AtomicInteger()
+  val getFileStatusCalls = new AtomicInteger()
+  def reset(): Unit = {
+    listStatusCalls.set(0)
+    getFileStatusCalls.set(0)
+  }
+}