feat(iceberg): native Iceberg scans of HDFS-backed tables via iceberg-rust's hdfs-native backend

.github/workflows/pr_build_linux.yml

@@ -538,6 +538,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

.github/workflows/pr_build_macos.yml

@@ -173,6 +173,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

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

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.
 

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
 

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,47 @@ 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. Of the session's
+Hadoop configuration (the driver's XML and `spark.hadoop.*`), Comet forwards the settings the
+client reads: `dfs.client.use.datanode.hostname`, `dfs.client.socket-timeout`,
+`dfs.client.failover.max.attempts`, `dfs.client.failover.sleep.base.millis` and
+`dfs.client.failover.sleep.max.millis` when the session sets them, and the NameNodes of 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 catalog property under `hadoop.` sets any client key directly and wins over the
+forwarded value. The client does not read the JVM client's other settings, among them short-circuit
+and hedged reads, `dfs.client.retry.*`, `ipc.client.*`, `hadoop.rpc.protection` and the
+per-nameservice `dfs.client.failover.*.<ns>` keys. Data transfer is encrypted when the NameNode
+requires it (`dfs.encrypt.data.transfer`), as for 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. A listing is kept per
+table for an hour, or for `spark.sql.metadataCacheTTLSeconds` when that is set, so a table
+re-created at the same path lists again on its first scan, and a new partition lists its parent
+again; a scan that would list more than a million directory entries falls back, and later scans
+continue from the listings kept, which hold a million entries at most.
+
+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, when the application is
+submitted (Comet reads it from the SparkContext's configuration; a value set on a session
+afterwards never reaches the executors). 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.
+To read HDFS tables with Spark's Iceberg reader and keep the native scan for other storage, set
+`spark.comet.scan.icebergNative.hdfs.enabled=false`. Writes to HDFS use Iceberg's own writer.
+
 ### Current limitations
 
 The following scenarios will fall back to the JVM Iceberg reader:
@@ -210,6 +250,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
@@ -793,16 +839,19 @@ the [S3](datasources.md#s3-limitations) and [Azure](datasources.md#azure-limitat
 
 Which entries apply depends on which native paths are enabled:
 
-| Native path    | Setting                                  | Default |
-| -------------- | ---------------------------------------- | ------- |
-| Iceberg scan   | `spark.comet.scan.icebergNative.enabled` | `true`  |
-| Iceberg writer | `spark.comet.write.iceberg.enabled`      | `false` |
+| Native path          | Setting                                       | Default |
+| -------------------- | --------------------------------------------- | ------- |
+| Iceberg scan         | `spark.comet.scan.icebergNative.enabled`      | `true`  |
+| Iceberg scan of HDFS | `spark.comet.scan.icebergNative.hdfs.enabled` | `true`  |
+| Iceberg writer       | `spark.comet.write.iceberg.enabled`           | `false` |
 
 ### 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. A read of only some
-  of a struct's fields fails instead
+- **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